feat: Add CreateVectorIndexReference WriterTarget and SPI methods (#27261)

presto-analyzer/src/main/java/com/facebook/presto/sql/analyzer/Analysis.java

@@ -186,6 +186,9 @@ public class Analysis
     private Optional<TableHandle> callTarget = Optional.empty();
     private Optional<QuerySpecification> targetQuery = Optional.empty();
 
+    // for create vector index
+    private Optional<CreateVectorIndexAnalysis> createVectorIndexAnalysis = Optional.empty();
+
     // for create table
     private Optional<QualifiedObjectName> createTableDestination = Optional.empty();
     private Map<String, Expression> createTableProperties = ImmutableMap.of();
@@ -700,6 +703,16 @@ public Optional<QualifiedObjectName> getCreateTableDestination()
         return createTableDestination;
     }
 
+    public void setCreateVectorIndexAnalysis(CreateVectorIndexAnalysis analysis)
+    {
+        this.createVectorIndexAnalysis = Optional.of(analysis);
+    }
+
+    public Optional<CreateVectorIndexAnalysis> getCreateVectorIndexAnalysis()
+    {
+        return createVectorIndexAnalysis;
+    }
+
     public Optional<QualifiedObjectName> getProcedureName()
     {
         return procedureName;
@@ -1937,4 +1950,53 @@ public Scope getTargetTableScope()
             return targetTableScope;
         }
     }
+
+    @Immutable
+    public static final class CreateVectorIndexAnalysis
+    {
+        private final QualifiedObjectName sourceTableName;
+        private final QualifiedObjectName targetTableName;
+        private final List<Identifier> columns;
+        private final Map<String, Expression> properties;
+        private final Optional<Expression> updatingFor;
+
+        public CreateVectorIndexAnalysis(
+                QualifiedObjectName sourceTableName,
+                QualifiedObjectName targetTableName,
+                List<Identifier> columns,
+                Map<String, Expression> properties,
+                Optional<Expression> updatingFor)
+        {
+            this.sourceTableName = requireNonNull(sourceTableName, "sourceTableName is null");
+            this.targetTableName = requireNonNull(targetTableName, "targetTableName is null");
+            this.columns = ImmutableList.copyOf(requireNonNull(columns, "columns is null"));
+            this.properties = ImmutableMap.copyOf(requireNonNull(properties, "properties is null"));
+            this.updatingFor = requireNonNull(updatingFor, "updatingFor is null");
+        }
+
+        public QualifiedObjectName getSourceTableName()
+        {
+            return sourceTableName;
+        }
+
+        public QualifiedObjectName getTargetTableName()
+        {
+            return targetTableName;
+        }
+
+        public List<Identifier> getColumns()
+        {
+            return columns;
+        }
+
+        public Map<String, Expression> getProperties()
+        {
+            return properties;
+        }
+
+        public Optional<Expression> getUpdatingFor()
+        {
+            return updatingFor;
+        }
+    }
 }

presto-analyzer/src/main/java/com/facebook/presto/sql/analyzer/utils/StatementUtils.java

@@ -30,6 +30,7 @@
 import com.facebook.presto.sql.tree.CreateTableAsSelect;
 import com.facebook.presto.sql.tree.CreateTag;
 import com.facebook.presto.sql.tree.CreateType;
+import com.facebook.presto.sql.tree.CreateVectorIndex;
 import com.facebook.presto.sql.tree.CreateView;
 import com.facebook.presto.sql.tree.Deallocate;
 import com.facebook.presto.sql.tree.Delete;
@@ -107,6 +108,7 @@ private StatementUtils() {}
         builder.put(CreateTableAsSelect.class, QueryType.INSERT);
         builder.put(Insert.class, QueryType.INSERT);
         builder.put(RefreshMaterializedView.class, QueryType.INSERT);
+        builder.put(CreateVectorIndex.class, QueryType.INSERT);
 
         builder.put(Delete.class, QueryType.DELETE);
         builder.put(Update.class, QueryType.UPDATE);

presto-main-base/src/main/java/com/facebook/presto/sql/analyzer/StatementAnalyzer.java

@@ -110,6 +110,7 @@
 import com.facebook.presto.sql.tree.CreateSchema;
 import com.facebook.presto.sql.tree.CreateTable;
 import com.facebook.presto.sql.tree.CreateTableAsSelect;
+import com.facebook.presto.sql.tree.CreateVectorIndex;
 import com.facebook.presto.sql.tree.CreateView;
 import com.facebook.presto.sql.tree.Cube;
 import com.facebook.presto.sql.tree.Deallocate;
@@ -1144,6 +1145,71 @@ protected Scope visitCreateTable(CreateTable node, Optional<Scope> scope)
             return createAndAssignScope(node, scope);
         }
 
+        @Override
+        protected Scope visitCreateVectorIndex(CreateVectorIndex node, Optional<Scope> scope)
+        {
+            QualifiedObjectName sourceTableName = createQualifiedObjectName(session, node, node.getTableName(), metadata);
+            if (!metadataResolver.tableExists(sourceTableName)) {
+                throw new SemanticException(MISSING_TABLE, node, "Source table '%s' does not exist", sourceTableName);
+            }
+
+            QualifiedObjectName targetTable = createQualifiedObjectName(session, node, node.getIndexName(), metadata);
+            if (metadataResolver.tableExists(targetTable)) {
+                throw new SemanticException(TABLE_ALREADY_EXISTS, node, "Destination table '%s' already exists", targetTable);
+            }
+
+            // Analyze the source table to build a proper scope with typed columns
+            // Use AllowAllAccessControl since we check permissions separately below
+            StatementAnalyzer analyzer = new StatementAnalyzer(
+                    analysis,
+                    metadata,
+                    sqlParser,
+                    new AllowAllAccessControl(),
+                    session,
+                    warningCollector);
+
+            Table sourceTable = new Table(node.getTableName());
+            Scope tableScope = analyzer.analyze(sourceTable, scope);
+
+            // Validate that specified columns exist in the source table
+            TableHandle sourceTableHandle = metadataResolver.getTableHandle(sourceTableName).get();
+            Map<String, ColumnHandle> sourceColumns = metadataResolver.getColumnHandles(sourceTableHandle);
+            for (Identifier column : node.getColumns()) {
+                if (!sourceColumns.containsKey(column.getValue())) {
+                    throw new SemanticException(MISSING_COLUMN, column, "Column '%s' does not exist in source table '%s'", column.getValue(), sourceTableName);
+                }
+            }
+
+            // Analyze UPDATING FOR predicate (validates column references, types, etc.)
+            node.getUpdatingFor().ifPresent(where -> analyzeWhere(node, tableScope, where));
+
+            validateProperties(node.getProperties(), scope);
+
+            Map<String, Expression> allProperties = mapFromProperties(node.getProperties());
+
+            // user must have read permission on the source table to create a vector index
+            Multimap<QualifiedObjectName, Subfield> tableColumnMap = ImmutableMultimap.<QualifiedObjectName, Subfield>builder()
+                    .putAll(sourceTableName, sourceColumns.keySet().stream()
+                            .map(column -> new Subfield(column, ImmutableList.of()))
+                            .collect(toImmutableSet()))
+                    .build();
+            analysis.addTableColumnAndSubfieldReferences(accessControl, session.getIdentity(),
+                    session.getTransactionId(), session.getAccessControlContext(), tableColumnMap, tableColumnMap);
+
+            analysis.addAccessControlCheckForTable(TABLE_CREATE,
+                    new AccessControlInfoForTable(accessControl, session.getIdentity(),
+                            session.getTransactionId(), session.getAccessControlContext(), targetTable));
+
+            analysis.setCreateVectorIndexAnalysis(new Analysis.CreateVectorIndexAnalysis(
+                    sourceTableName,
+                    targetTable,
+                    node.getColumns(),
+                    allProperties,
+                    node.getUpdatingFor()));
+
+            return createAndAssignScope(node, scope, Field.newUnqualified(node.getLocation(), "result", VARCHAR));
+        }
+
         @Override
         protected Scope visitProperty(Property node, Optional<Scope> scope)
         {

presto-main-base/src/test/java/com/facebook/presto/sql/analyzer/TestAnalyzer.java

@@ -82,6 +82,7 @@
 import static com.facebook.presto.sql.analyzer.SemanticErrorCode.SAMPLE_PERCENTAGE_OUT_OF_RANGE;
 import static com.facebook.presto.sql.analyzer.SemanticErrorCode.SCHEMA_NOT_SPECIFIED;
 import static com.facebook.presto.sql.analyzer.SemanticErrorCode.STANDALONE_LAMBDA;
+import static com.facebook.presto.sql.analyzer.SemanticErrorCode.TABLE_ALREADY_EXISTS;
 import static com.facebook.presto.sql.analyzer.SemanticErrorCode.TABLE_FUNCTION_COLUMN_NOT_FOUND;
 import static com.facebook.presto.sql.analyzer.SemanticErrorCode.TABLE_FUNCTION_DUPLICATE_RANGE_VARIABLE;
 import static com.facebook.presto.sql.analyzer.SemanticErrorCode.TABLE_FUNCTION_IMPLEMENTATION_ERROR;
@@ -2399,4 +2400,48 @@ public void testInvalidMerge()
         assertFails(NOT_SUPPORTED, "line 1:1: Merging into materialized views is not supported",
                 "MERGE INTO mv1 USING t1 ON mv1.a = t1.a WHEN MATCHED THEN  UPDATE SET id = bar.id + 1");
     }
+
+    @Test
+    public void testCreateVectorIndex()
+    {
+        // basic success cases
+        analyze("CREATE VECTOR INDEX test_index ON t1(a, b)");
+        analyze("CREATE VECTOR INDEX test_index ON t1(a, b) WITH (p1 = 'val1')");
+        analyze("CREATE VECTOR INDEX test_index ON t1(a, b) WITH (p1 = 'val1', p2 = 'val2')");
+
+        // with UPDATING FOR clause
+        analyze("CREATE VECTOR INDEX test_index ON t1(a, b) UPDATING FOR a > 10");
+        analyze("CREATE VECTOR INDEX test_index ON t1(a, b) WITH (p1 = 'val1') UPDATING FOR a BETWEEN 1 AND 100");
+
+        // single column
+        analyze("CREATE VECTOR INDEX test_index ON t1(a)");
+
+        // source table does not exist
+        assertFails(MISSING_TABLE, ".*Source table '.*' does not exist",
+                "CREATE VECTOR INDEX test_index ON nonexistent_table(a, b)");
+
+        // destination table already exists (using an existing table name as the index name)
+        assertFails(TABLE_ALREADY_EXISTS, ".*already exists",
+                "CREATE VECTOR INDEX t1 ON t2(a, b)");
+
+        // column does not exist in source table
+        assertFails(MISSING_COLUMN, ".*Column 'unknown' does not exist in source table '.*'",
+                "CREATE VECTOR INDEX test_index ON t1(a, unknown)");
+        assertFails(MISSING_COLUMN, ".*Column 'nonexistent' does not exist in source table '.*'",
+                "CREATE VECTOR INDEX test_index ON t1(nonexistent)");
+
+        // duplicate properties
+        assertFails(DUPLICATE_PROPERTY, ".* Duplicate property: p1",
+                "CREATE VECTOR INDEX test_index ON t1(a, b) WITH (p1 = 'v1', p2 = 'v2', p1 = 'v3')");
+        assertFails(DUPLICATE_PROPERTY, ".* Duplicate property: p1",
+                "CREATE VECTOR INDEX test_index ON t1(a, b) WITH (p1 = 'v1', \"p1\" = 'v2')");
+
+        // unresolved property value
+        assertFails(MISSING_ATTRIBUTE, ".*'y' cannot be resolved",
+                "CREATE VECTOR INDEX test_index ON t1(a, b) WITH (p1 = y)");
+
+        // UPDATING FOR with invalid column reference
+        assertFails(MISSING_ATTRIBUTE, ".*",
+                "CREATE VECTOR INDEX test_index ON t1(a, b) UPDATING FOR nonexistent_col > 10");
+    }
 }

presto-spi/src/main/java/com/facebook/presto/spi/connector/ConnectorMetadata.java

@@ -502,6 +502,22 @@ default Optional<ConnectorOutputMetadata> finishCreateTable(ConnectorSession ses
         throw new PrestoException(GENERIC_INTERNAL_ERROR, "ConnectorMetadata beginCreateTable() is implemented without finishCreateTable()");
     }
 
+    /**
+     * Begin the atomic creation of a vector index with data.
+     */
+    default ConnectorOutputTableHandle beginCreateVectorIndex(ConnectorSession session, ConnectorTableMetadata indexMetadata, Optional<ConnectorNewTableLayout> layout, SchemaTableName sourceTableName)
+    {
+        throw new PrestoException(NOT_SUPPORTED, "This connector does not support creating vector indexes");
+    }
+
+    /**
+     * Finish a vector index creation with data after the data is written.
+     */
+    default Optional<ConnectorOutputMetadata> finishCreateVectorIndex(ConnectorSession session, ConnectorOutputTableHandle tableHandle, Collection<Slice> fragments, Collection<ComputedStatistics> computedStatistics)
+    {
+        throw new PrestoException(NOT_SUPPORTED, "This connector does not support creating vector indexes");
+    }
+
     /**
      * Start a SELECT/UPDATE/INSERT/DELETE query. This notification is triggered after the planning phase completes.
      */

presto-spi/src/main/java/com/facebook/presto/spi/connector/classloader/ClassLoaderSafeConnectorMetadata.java

@@ -450,6 +450,22 @@ public Optional<ConnectorOutputMetadata> finishCreateTable(ConnectorSession sess
         }
     }
 
+    @Override
+    public ConnectorOutputTableHandle beginCreateVectorIndex(ConnectorSession session, ConnectorTableMetadata indexMetadata, Optional<ConnectorNewTableLayout> layout, SchemaTableName sourceTableName)
+    {
+        try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
+            return delegate.beginCreateVectorIndex(session, indexMetadata, layout, sourceTableName);
+        }
+    }
+
+    @Override
+    public Optional<ConnectorOutputMetadata> finishCreateVectorIndex(ConnectorSession session, ConnectorOutputTableHandle tableHandle, Collection<Slice> fragments, Collection<ComputedStatistics> computedStatistics)
+    {
+        try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
+            return delegate.finishCreateVectorIndex(session, tableHandle, fragments, computedStatistics);
+        }
+    }
+
     @Override
     public void beginQuery(ConnectorSession session)
     {

presto-spi/src/main/java/com/facebook/presto/spi/plan/TableWriterNode.java

@@ -698,4 +698,67 @@ public String toString()
             return procedureName.toString();
         }
     }
+
+    public static class CreateVectorIndexReference
+            extends WriterTarget
+    {
+        private final ConnectorId connectorId;
+        private final ConnectorTableMetadata tableMetadata;
+        private final Optional<NewTableLayout> layout;
+        private final Optional<List<OutputColumnMetadata>> columns;
+        private final SchemaTableName sourceTableName;
+
+        public CreateVectorIndexReference(
+                ConnectorId connectorId,
+                ConnectorTableMetadata tableMetadata,
+                Optional<NewTableLayout> layout,
+                Optional<List<OutputColumnMetadata>> columns,
+                SchemaTableName sourceTableName)
+        {
+            this.connectorId = requireNonNull(connectorId, "connectorId is null");
+            this.tableMetadata = requireNonNull(tableMetadata, "tableMetadata is null");
+            this.layout = requireNonNull(layout, "layout is null");
+            this.columns = requireNonNull(columns, "columns is null");
+            this.sourceTableName = requireNonNull(sourceTableName, "sourceTableName is null");
+        }
+
+        @Override
+        public ConnectorId getConnectorId()
+        {
+            return connectorId;
+        }
+
+        public ConnectorTableMetadata getTableMetadata()
+        {
+            return tableMetadata;
+        }
+
+        public Optional<NewTableLayout> getLayout()
+        {
+            return layout;
+        }
+
+        @Override
+        public SchemaTableName getSchemaTableName()
+        {
+            return tableMetadata.getTable();
+        }
+
+        @Override
+        public String toString()
+        {
+            return connectorId + "." + tableMetadata.getTable();
+        }
+
+        @Override
+        public Optional<List<OutputColumnMetadata>> getOutputColumns()
+        {
+            return columns;
+        }
+
+        public SchemaTableName getSourceTableName()
+        {
+            return sourceTableName;
+        }
+    }
 }