refactor(isthmus): derive CREATE-statement table types from the provider

isthmus/src/main/java/io/substrait/isthmus/sql/SubstraitCreateStatementParser.java

@@ -1,8 +1,6 @@
 package io.substrait.isthmus.sql;
 
 import io.substrait.isthmus.ConverterProvider;
-import io.substrait.isthmus.SqlConverterBase;
-import io.substrait.isthmus.SubstraitTypeSystem;
 import io.substrait.isthmus.Utils;
 import io.substrait.isthmus.calcite.SubstraitTable;
 import java.util.ArrayList;
@@ -11,6 +9,7 @@
 import org.apache.calcite.jdbc.CalciteSchema;
 import org.apache.calcite.prepare.CalciteCatalogReader;
 import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlNodeList;
 import org.apache.calcite.sql.ddl.SqlColumnDeclaration;
@@ -25,19 +24,18 @@
 /** Utility class for parsing CREATE statements into a {@link CalciteCatalogReader} */
 public class SubstraitCreateStatementParser {
 
-  /** An empty catalog reader used for validating CREATE statements. */
+  /**
+   * An empty catalog reader used for validating CREATE statements, configured from {@link
+   * ConverterProvider#DEFAULT}.
+   */
   public static final CalciteCatalogReader EMPTY_CATALOG =
-      new CalciteCatalogReader(
-          CalciteSchema.createRootSchema(false),
-          List.of(),
-          SubstraitTypeSystem.TYPE_FACTORY,
-          SqlConverterBase.CONNECTION_CONFIG);
-
-  /** SQL validator configured for validating CREATE statements against the empty catalog. */
-  public static final SqlValidator VALIDATOR =
-      new SubstraitSqlValidator(
-          // as we are validating CREATE statements, an empty catalog suffices
-          EMPTY_CATALOG);
+      createEmptyCatalog(ConverterProvider.DEFAULT);
+
+  /**
+   * SQL validator configured for validating CREATE statements against the empty catalog, using
+   * {@link ConverterProvider#DEFAULT}.
+   */
+  public static final SqlValidator VALIDATOR = createValidator(ConverterProvider.DEFAULT);
 
   /**
    * Parses a SQL string containing only CREATE statements into a list of {@link SubstraitTable}s.
@@ -71,6 +69,7 @@ public static List<SubstraitTable> processCreateStatements(
       @NonNull final ConverterProvider converterProvider, @NonNull final String createStatements)
       throws SqlParseException {
     final List<SubstraitTable> tableList = new ArrayList<>();
+    final SqlValidator validator = createValidator(converterProvider);
 
     final List<SqlNode> sqlNode =
         SubstraitSqlStatementParser.parseStatements(createStatements, converterProvider);
@@ -89,7 +88,12 @@ public static List<SubstraitTable> processCreateStatements(
         throw fail("CTAS not supported.", create.name.getParserPosition());
       }
 
-      tableList.add(createSubstraitTable(create.name.names.get(0), create.columnList));
+      tableList.add(
+          createSubstraitTable(
+              converterProvider.getTypeFactory(),
+              validator,
+              create.name.names.get(0),
+              create.columnList));
     }
 
     return tableList;
@@ -200,6 +204,7 @@ private static CalciteSchema processCreateStatementsToSchema(
       @NonNull final ConverterProvider converterProvider, @NonNull final String... createStatements)
       throws SqlParseException {
     final CalciteSchema rootSchema = CalciteSchema.createRootSchema(false);
+    final SqlValidator validator = createValidator(converterProvider);
 
     for (final String statement : createStatements) {
       final List<SqlNode> sqlNode =
@@ -219,7 +224,10 @@ private static CalciteSchema processCreateStatementsToSchema(
         final String tableName = names.get(names.size() - 1);
         final CalciteSchema.TableEntry table = schema.getTable(tableName, false);
         if (table == null) {
-          schema.add(tableName, createSubstraitTable(tableName, create.columnList));
+          schema.add(
+              tableName,
+              createSubstraitTable(
+                  converterProvider.getTypeFactory(), validator, tableName, create.columnList));
         } else {
           throw fail("Table must not be defined more than once", parsed.getParserPosition());
         }
@@ -233,14 +241,19 @@ private static CalciteSchema processCreateStatementsToSchema(
    * Creates a new {@link SubstraitTable} with the given table name and the table schema from the
    * given {@link SqlNodeList} containing {@link SqlColumnDeclaration}s.
    *
+   * @param typeFactory the type factory used to build the table's row type; must not be null
+   * @param validator the validator used to derive the column types; must not be null
    * @param tableName the table name to use; must not be null
    * @param columnList the {@link SqlNodeList} containing {@link SqlColumnDeclaration}s to build the
    *     table schema from; must not be null
    * @return the constructed {@link SubstraitTable}
    * @throws SqlParseException if the column list contains unexpected nodes or invalid names
    */
   private static SubstraitTable createSubstraitTable(
-      @NonNull final String tableName, @NonNull final SqlNodeList columnList)
+      @NonNull final RelDataTypeFactory typeFactory,
+      @NonNull final SqlValidator validator,
+      @NonNull final String tableName,
+      @NonNull final SqlNodeList columnList)
       throws SqlParseException {
     final List<String> names = new ArrayList<>();
     final List<RelDataType> columnTypes = new ArrayList<>();
@@ -263,10 +276,40 @@ private static SubstraitTable createSubstraitTable(
       }
 
       names.add(col.name.names.get(0));
-      columnTypes.add(col.dataType.deriveType(VALIDATOR));
+      columnTypes.add(col.dataType.deriveType(validator));
     }
 
-    return new SubstraitTable(
-        tableName, SubstraitTypeSystem.TYPE_FACTORY.createStructType(columnTypes, names));
+    return new SubstraitTable(tableName, typeFactory.createStructType(columnTypes, names));
+  }
+
+  /**
+   * Creates an empty catalog reader for validating CREATE statements, using the type factory and
+   * connection configuration from the given {@link ConverterProvider}.
+   *
+   * @param converterProvider the converter provider supplying the type factory and connection
+   *     configuration; must not be null
+   * @return an empty {@link CalciteCatalogReader}
+   */
+  private static CalciteCatalogReader createEmptyCatalog(
+      @NonNull final ConverterProvider converterProvider) {
+    return new CalciteCatalogReader(
+        CalciteSchema.createRootSchema(false),
+        List.of(),
+        converterProvider.getTypeFactory(),
+        converterProvider.getCalciteConnectionConfig());
+  }
+
+  /**
+   * Creates a SQL validator for deriving the column types of CREATE statements, configured from the
+   * given {@link ConverterProvider}. As only CREATE statements are validated, an empty catalog
+   * suffices.
+   *
+   * @param converterProvider the converter provider supplying the validator configuration; must not
+   *     be null
+   * @return a {@link SqlValidator} for validating CREATE statements
+   */
+  private static SqlValidator createValidator(@NonNull final ConverterProvider converterProvider) {
+    return new SubstraitSqlValidator(
+        createEmptyCatalog(converterProvider), converterProvider.getSqlOperatorTable());
   }
 }

isthmus/src/test/java/io/substrait/isthmus/sql/CreateStatementParserProviderConfigTest.java

@@ -0,0 +1,141 @@
+package io.substrait.isthmus.sql;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.substrait.isthmus.ConverterProvider;
+import io.substrait.isthmus.SubstraitTypeSystem;
+import io.substrait.isthmus.calcite.SubstraitTable;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.calcite.prepare.CalciteCatalogReader;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.parser.SqlParseException;
+import org.apache.calcite.sql.type.SqlTypeFactoryImpl;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Verifies that the CREATE-statement parser sources its type factory and connection configuration
+ * from the injected {@link ConverterProvider} rather than from global defaults, for both the
+ * catalog-reader and the {@link SubstraitTable} entry points.
+ */
+class CreateStatementParserProviderConfigTest {
+
+  private static final String CREATE_STATEMENT =
+      "CREATE TABLE employees (id BIGINT, name VARCHAR, salary DECIMAL(10, 2))";
+
+  /**
+   * A type factory that records the row types it is asked to build and the SQL types it is asked to
+   * create, so a test can tell which factory produced a table's row type and which one the
+   * validator derived its column types through.
+   *
+   * <p>Recording the requests is what makes the column-derivation check meaningful: {@code
+   * RelDataTypeFactoryImpl.canonize} interns through a {@code static} cache shared by every factory
+   * instance, so comparing interned type instances cannot distinguish one factory from another.
+   */
+  static final class RecordingTypeFactory extends SqlTypeFactoryImpl {
+
+    private boolean createStructTypeCalled;
+    private final List<SqlTypeName> requestedSqlTypeNames = new ArrayList<>();
+
+    RecordingTypeFactory() {
+      super(SubstraitTypeSystem.TYPE_SYSTEM);
+    }
+
+    @Override
+    public RelDataType createStructType(
+        final List<RelDataType> typeList, final List<String> fieldNameList) {
+      createStructTypeCalled = true;
+      return super.createStructType(typeList, fieldNameList);
+    }
+
+    @Override
+    public RelDataType createSqlType(final SqlTypeName typeName) {
+      requestedSqlTypeNames.add(typeName);
+      return super.createSqlType(typeName);
+    }
+
+    @Override
+    public RelDataType createSqlType(final SqlTypeName typeName, final int precision) {
+      requestedSqlTypeNames.add(typeName);
+      return super.createSqlType(typeName, precision);
+    }
+
+    @Override
+    public RelDataType createSqlType(
+        final SqlTypeName typeName, final int precision, final int scale) {
+      requestedSqlTypeNames.add(typeName);
+      return super.createSqlType(typeName, precision, scale);
+    }
+
+    boolean wasUsedForStructType() {
+      return createStructTypeCalled;
+    }
+
+    List<SqlTypeName> requestedSqlTypeNames() {
+      return List.copyOf(requestedSqlTypeNames);
+    }
+  }
+
+  @Test
+  void catalogPathUsesProviderTypeFactory() throws SqlParseException {
+    RecordingTypeFactory typeFactory = new RecordingTypeFactory();
+    ConverterProvider provider = ConverterProvider.builder().typeFactory(typeFactory).build();
+
+    CalciteCatalogReader catalog =
+        SubstraitCreateStatementParser.processCreateStatementsToCatalog(provider, CREATE_STATEMENT);
+
+    assertSame(typeFactory, catalog.getTypeFactory());
+    assertTrue(
+        typeFactory.wasUsedForStructType(),
+        "the provider's type factory should build the table row type");
+  }
+
+  @Test
+  void tableListPathUsesProviderTypeFactory() throws SqlParseException {
+    RecordingTypeFactory typeFactory = new RecordingTypeFactory();
+    ConverterProvider provider = ConverterProvider.builder().typeFactory(typeFactory).build();
+
+    List<SubstraitTable> tables =
+        SubstraitCreateStatementParser.processCreateStatements(provider, CREATE_STATEMENT);
+
+    assertEquals(1, tables.size());
+    assertTrue(
+        typeFactory.wasUsedForStructType(),
+        "the provider's type factory should build the table row type");
+  }
+
+  /**
+   * The validator that derives the column types is built from the provider, so the declared column
+   * types are requested from the provider's type factory — covering the validator path, not just
+   * the enclosing struct type.
+   */
+  @Test
+  void columnTypesAreDerivedWithProviderTypeFactory() throws SqlParseException {
+    RecordingTypeFactory typeFactory = new RecordingTypeFactory();
+    ConverterProvider provider = ConverterProvider.builder().typeFactory(typeFactory).build();
+
+    SubstraitCreateStatementParser.processCreateStatementsToCatalog(provider, CREATE_STATEMENT);
+
+    assertTrue(
+        typeFactory
+            .requestedSqlTypeNames()
+            .containsAll(List.of(SqlTypeName.BIGINT, SqlTypeName.VARCHAR, SqlTypeName.DECIMAL)),
+        "the validator should derive the declared column types through the provider's type "
+            + "factory, but it only saw "
+            + typeFactory.requestedSqlTypeNames());
+  }
+
+  /** The default entry points keep working off {@link ConverterProvider#DEFAULT}. */
+  @Test
+  void defaultPathStillUsesSystemDefaults() throws SqlParseException {
+    CalciteCatalogReader catalog =
+        SubstraitCreateStatementParser.processCreateStatementsToCatalog(CREATE_STATEMENT);
+
+    assertSame(SubstraitTypeSystem.TYPE_FACTORY, catalog.getTypeFactory());
+    assertFalse(catalog.nameMatcher().isCaseSensitive(), "default config is case-insensitive");
+  }
+}