voonhous commented on code in PR #20012: URL: https://github.com/apache/hudi/pull/20012#discussion_r4090410342
########## hudi-trino/src/main/java/io/trino/plugin/hudi/HudiTableInitializer.java: ########## @@ -0,0 +1,116 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hudi; + +import io.trino.filesystem.TrinoFileSystem; +import io.trino.plugin.hudi.storage.TrinoStorageConfiguration; +import io.trino.spi.TrinoException; +import io.trino.spi.connector.ConnectorTableMetadata; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.HoodieTableVersion; +import org.apache.hudi.storage.StoragePath; + +import java.io.IOException; +import java.util.List; +import java.util.Map; + +import static io.trino.plugin.hudi.HudiErrorCode.HUDI_META_CLIENT_ERROR; +import static io.trino.plugin.hudi.HudiTableProperties.getHiveStylePartitioning; +import static io.trino.plugin.hudi.HudiTableProperties.getHoodieProperties; +import static io.trino.plugin.hudi.HudiTableProperties.getKeyGeneratorClass; +import static io.trino.plugin.hudi.HudiTableProperties.getOrderingFields; +import static io.trino.plugin.hudi.HudiTableProperties.getPartitionedBy; +import static io.trino.plugin.hudi.HudiTableProperties.getPrimaryKey; +import static io.trino.plugin.hudi.HudiTableProperties.getRecordMergeMode; +import static io.trino.plugin.hudi.HudiTableProperties.getTableType; +import static java.lang.String.format; + +/** + * Writes a table's {@code .hoodie} directory. Pure hudi-common: no Spark, no write client and no + * metastore, so it can run before the catalog entry exists without entering any row-writing path. + */ +public final class HudiTableInitializer +{ + /** + * The {@code hoodie.table.version} written by this connector. + * <p> + * Pinned rather than inherited from {@link HoodieTableVersion#current()} so that a future bump of + * {@code current()} cannot silently change what the connector emits. Raising this should be a + * deliberate change with a test diff, not a side effect of upgrading Hudi; the assertion in + * {@code TestHudiDdl} fails when the two drift. + */ + public static final HoodieTableVersion CREATED_TABLE_VERSION = HoodieTableVersion.TEN; + + private HudiTableInitializer() {} + + /** + * Initializes storage for an empty table. + * + * @param tableSchema the schema that also produces the metastore column list, so the two cannot + * disagree + */ + public static void initializeTable( + TrinoFileSystem fileSystem, + String basePath, + ConnectorTableMetadata tableMetadata, + HoodieSchema tableSchema) + { + Map<String, Object> properties = tableMetadata.getProperties(); + HoodieTableMetaClient.TableBuilder builder = HoodieTableMetaClient.newTableBuilder() + .setTableType(getTableType(properties)) + .setTableName(tableMetadata.getTable().getTableName()) + .setDatabaseName(tableMetadata.getTable().getSchemaName()) + .setTableVersion(CREATED_TABLE_VERSION) + .setTableCreateSchema(tableSchema.toAvroSchema().toString()); + + List<String> primaryKey = getPrimaryKey(properties); + if (!primaryKey.isEmpty()) { + builder.setRecordKeyFields(String.join(",", primaryKey)); + } + List<String> partitionedBy = getPartitionedBy(properties); + if (!partitionedBy.isEmpty()) { + builder.setPartitionFields(String.join(",", partitionedBy)); + } + List<String> orderingFields = getOrderingFields(properties); + if (!orderingFields.isEmpty()) { + builder.setOrderingFields(String.join(",", orderingFields)); + } + // Each of the following is written only when the user asked for it. Left unset, Hudi applies + // its own default at write time, which keeps a Trino-created table indistinguishable from + // one another engine created with the same DDL. + getRecordMergeMode(properties).ifPresent(builder::setRecordMergeMode); + getKeyGeneratorClass(properties).ifPresent(builder::setKeyGeneratorClassProp); Review Comment: **major:** When `key_generator_class` is unset nothing is persisted, and no later writer fills it in (`HoodieSparkSqlWriter` only builds table config for a new table). Spark SQL then falls back to `ComplexKeyGenerator` in `SqlKeyGenerator` (keys like `id:1` at v9+, HUDI-9666), while a DataFrame write infers `SimpleKeyGenerator` (keys like `1`), and `HoodieWriterUtils` skips the mismatch check when the table key generator is null. Mixing the two writers would duplicate rows (traced, not executed). Spark SQL `CREATE TABLE` always pins one (`HoodieCatalogTable.scala:343-357`). Could we infer the type the way `KeyGenUtils.inferKeyGeneratorType` does, persist it via `setKeyGeneratorType`, and assert it in `TestHudiDdl`? ########## hudi-trino/src/main/java/io/trino/plugin/hudi/HudiMetadata.java: ########## @@ -304,6 +321,202 @@ public Optional<Object> getInfo(ConnectorSession session, ConnectorTableHandle t return Optional.of(new HudiTableInfo(table.getSchemaTableName(), table.getTableType().name(), table.getBasePath())); } + /** + * Creates an empty table: its {@code .hoodie} directory on storage and its catalog entry. + * <p> + * This is the single-call path only. {@code beginCreateTable}/{@code finishCreateTable} and a + * page sink are deliberately absent, so {@code CREATE TABLE AS} and {@code INSERT} still fail as + * unsupported; no rows are written here. + * <p> + * An explicit {@code location} makes the table external, matching Hudi's Spark SQL behaviour. An + * omitted one makes it managed, at {@code <schemaLocation>/<tableName>}, which is what decides + * whether {@code DROP TABLE} later deletes the data. + */ + @Override + public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, SaveMode saveMode) + { + SchemaTableName schemaTableName = tableMetadata.getTable(); + if (saveMode == SaveMode.REPLACE) { + // Replacing a table means deciding what happens to the rows already in it, which is the + // write path this connector does not yet have. + throw new TrinoException(NOT_SUPPORTED, "This connector does not support replacing tables"); + } + Database database = metastore.getDatabase(schemaTableName.getSchemaName()) + .orElseThrow(() -> new SchemaNotFoundException(schemaTableName.getSchemaName())); + if (metastore.getTable(schemaTableName.getSchemaName(), schemaTableName.getTableName()).isPresent()) { + if (saveMode == SaveMode.IGNORE) { + return; + } + throw new TrinoException(ALREADY_EXISTS, "Table already exists: " + schemaTableName); + } + + // Everything that can be rejected is rejected before storage is touched, so a bad statement + // leaves nothing behind. + HudiTableValidation.validateCreateTable(tableMetadata); + + Map<String, Object> properties = tableMetadata.getProperties(); + Optional<String> explicitLocation = getTableLocation(properties); + boolean external = explicitLocation.isPresent(); + String basePath = explicitLocation.orElseGet(() -> defaultTableLocation(database, schemaTableName)); + + TrinoFileSystem fileSystem = fileSystemFactory.create(session); + checkLocationIsEmpty(fileSystem, basePath); + + // One schema object produces both hoodie.table.create.schema and the metastore column list, + // so the two cannot disagree (HUDI-9435). + HoodieSchema tableSchema = HudiSchemaConverter.toTableSchema( + tableMetadata.getColumns(), schemaTableName.getTableName()); + Table table = HudiMetastoreTables.buildTable( + schemaTableName, + basePath, + getTableType(properties), + tableSchema, + getPartitionedBy(properties), + external, + Optional.of(session.getUser()), + tableMetadata.getComment()); + try { + HudiTableInitializer.initializeTable(fileSystem, basePath, tableMetadata, tableSchema); + metastore.createTable(table, NO_PRIVILEGES); + } + catch (TableAlreadyExistsException e) { + // Another CREATE TABLE may have initialized the same managed location and won the + // metastore race. Its catalog entry now owns .hoodie, so deleting it would corrupt the + // live table. A different location still belongs to this failed attempt and is safe to + // clean up. + if (!isTableRegisteredAtLocation(schemaTableName, basePath, e)) { + cleanupTableMetadata(fileSystem, basePath, e); + } + throw e; + } + catch (RuntimeException e) { + // Initialization may have written some or all of .hoodie, but no catalog entry from + // this call references it. Left there it would make a retry fail the emptiness check. + // + // Only .hoodie is removed, never the base path: the base path may have been created by + // someone else, and this connector did not create it. On object storage there is no + // directory to delete in any case -- deleteDirectory removes the objects under the + // prefix, which is exactly the set initTable wrote or may have partially written. + cleanupTableMetadata(fileSystem, basePath, e); Review Comment: **major:** The ownership check from the concurrent-create fix only covers the `TableAlreadyExistsException` branch; this generic branch still deletes `.hoodie` unconditionally. On HDFS/local, a losing concurrent CREATE (or a second external CREATE at the same location) fails inside `initTable`, because `HudiTrinoStorage.create` ignores `overwrite` and `HdfsOutputFile.create()` refuses to overwrite `hoodie.properties`. It lands here and deletes the winner's `.hoodie` (traced in code, not executed). Could we skip cleanup when the cause chain contains `FileAlreadyExistsException`, and add a test that fails `initTable` that way and asserts `.hoodie` survives? ########## hudi-trino/src/main/java/io/trino/plugin/hudi/HudiTableInitializer.java: ########## @@ -0,0 +1,116 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hudi; + +import io.trino.filesystem.TrinoFileSystem; +import io.trino.plugin.hudi.storage.TrinoStorageConfiguration; +import io.trino.spi.TrinoException; +import io.trino.spi.connector.ConnectorTableMetadata; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.HoodieTableVersion; +import org.apache.hudi.storage.StoragePath; + +import java.io.IOException; +import java.util.List; +import java.util.Map; + +import static io.trino.plugin.hudi.HudiErrorCode.HUDI_META_CLIENT_ERROR; +import static io.trino.plugin.hudi.HudiTableProperties.getHiveStylePartitioning; +import static io.trino.plugin.hudi.HudiTableProperties.getHoodieProperties; +import static io.trino.plugin.hudi.HudiTableProperties.getKeyGeneratorClass; +import static io.trino.plugin.hudi.HudiTableProperties.getOrderingFields; +import static io.trino.plugin.hudi.HudiTableProperties.getPartitionedBy; +import static io.trino.plugin.hudi.HudiTableProperties.getPrimaryKey; +import static io.trino.plugin.hudi.HudiTableProperties.getRecordMergeMode; +import static io.trino.plugin.hudi.HudiTableProperties.getTableType; +import static java.lang.String.format; + +/** + * Writes a table's {@code .hoodie} directory. Pure hudi-common: no Spark, no write client and no + * metastore, so it can run before the catalog entry exists without entering any row-writing path. + */ +public final class HudiTableInitializer +{ + /** + * The {@code hoodie.table.version} written by this connector. + * <p> + * Pinned rather than inherited from {@link HoodieTableVersion#current()} so that a future bump of + * {@code current()} cannot silently change what the connector emits. Raising this should be a + * deliberate change with a test diff, not a side effect of upgrading Hudi; the assertion in + * {@code TestHudiDdl} fails when the two drift. + */ + public static final HoodieTableVersion CREATED_TABLE_VERSION = HoodieTableVersion.TEN; + + private HudiTableInitializer() {} + + /** + * Initializes storage for an empty table. + * + * @param tableSchema the schema that also produces the metastore column list, so the two cannot + * disagree + */ + public static void initializeTable( + TrinoFileSystem fileSystem, + String basePath, + ConnectorTableMetadata tableMetadata, + HoodieSchema tableSchema) + { + Map<String, Object> properties = tableMetadata.getProperties(); + HoodieTableMetaClient.TableBuilder builder = HoodieTableMetaClient.newTableBuilder() + .setTableType(getTableType(properties)) + .setTableName(tableMetadata.getTable().getTableName()) + .setDatabaseName(tableMetadata.getTable().getSchemaName()) + .setTableVersion(CREATED_TABLE_VERSION) + .setTableCreateSchema(tableSchema.toAvroSchema().toString()); + + List<String> primaryKey = getPrimaryKey(properties); + if (!primaryKey.isEmpty()) { + builder.setRecordKeyFields(String.join(",", primaryKey)); + } + List<String> partitionedBy = getPartitionedBy(properties); + if (!partitionedBy.isEmpty()) { + builder.setPartitionFields(String.join(",", partitionedBy)); + } + List<String> orderingFields = getOrderingFields(properties); + if (!orderingFields.isEmpty()) { + builder.setOrderingFields(String.join(",", orderingFields)); + } + // Each of the following is written only when the user asked for it. Left unset, Hudi applies + // its own default at write time, which keeps a Trino-created table indistinguishable from + // one another engine created with the same DDL. + getRecordMergeMode(properties).ifPresent(builder::setRecordMergeMode); + getKeyGeneratorClass(properties).ifPresent(builder::setKeyGeneratorClassProp); + getHiveStylePartitioning(properties).ifPresent(builder::setHiveStylePartitioningEnable); Review Comment: **minor:** The comment above says leaving these unset keeps the table indistinguishable from another engine's with the same DDL, but Spark SQL `CREATE TABLE` writes `hive_style_partitioning=true` and url-encode `false` (`HoodieCatalogTable.scala:340-341`), while here the unset key defaults to `false`. Same partitioned DDL, different directory layout; self-consistent, since every writer reads the same default. Not blocking, but could we default `hive_style_partitioning` to `true` to match Spark SQL, or reword the comment to state the difference? ########## hudi-trino/src/main/java/io/trino/plugin/hudi/util/HudiSchemaConverter.java: ########## @@ -0,0 +1,283 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hudi.util; + +import com.google.common.collect.ImmutableList; +import io.trino.spi.TrinoException; +import io.trino.spi.connector.ColumnMetadata; +import io.trino.spi.type.ArrayType; +import io.trino.spi.type.CharType; +import io.trino.spi.type.DecimalType; +import io.trino.spi.type.MapType; +import io.trino.spi.type.RowType; +import io.trino.spi.type.TimeType; +import io.trino.spi.type.TimestampType; +import io.trino.spi.type.TimestampWithTimeZoneType; +import io.trino.spi.type.Type; +import io.trino.spi.type.VarbinaryType; +import io.trino.spi.type.VarcharType; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.schema.HoodieSchemaField; +import org.apache.hudi.common.schema.HoodieSchemaType; +import org.apache.hudi.common.schema.HoodieSchemaUtils; + +import java.util.HashSet; +import java.util.List; +import java.util.Locale; +import java.util.Set; + +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; +import static io.trino.spi.type.BigintType.BIGINT; +import static io.trino.spi.type.BooleanType.BOOLEAN; +import static io.trino.spi.type.DateType.DATE; +import static io.trino.spi.type.DoubleType.DOUBLE; +import static io.trino.spi.type.IntegerType.INTEGER; +import static io.trino.spi.type.RealType.REAL; +import static io.trino.spi.type.SmallintType.SMALLINT; +import static io.trino.spi.type.TinyintType.TINYINT; +import static io.trino.spi.type.UuidType.UUID; +import static java.lang.String.format; + +/** + * Converts a Trino column list into the {@link HoodieSchema} that becomes a table's + * {@code hoodie.table.create.schema}. + * <p> + * This direction did not previously exist in the connector: the read path only maps Hudi/Avro + * types into Trino types. The mapping here is the inverse of Trino's own + * {@code NativeLogicalTypesAvroTypeManager}, so a column created through this converter reads back + * as the same Trino type -- except where a mapping is deliberately widening, noted per case below. + * <p> + * Mappings that widen, so a column does not read back as the type it was declared with: + * <ul> + * <li>{@code TINYINT} and {@code SMALLINT} become Avro {@code int} and read back as + * {@code INTEGER}. Avro has no narrower integer, and Spark's Avro conversion widens the same + * way, so rejecting them would make the connector stricter than its peers for no gain.</li> + * <li>{@code VARCHAR(n)} becomes Avro {@code string} and reads back as unbounded {@code VARCHAR}. + * Avro strings carry no length bound; nothing enforces {@code n} once another engine writes.</li> + * <li>{@code TIMESTAMP(p)} for a {@code p} that is not exactly 3 or 6 rounds up to the next + * representable precision, since Avro offers only millisecond and microsecond logical types. + * The widening is lossless. Separately, the Hive Metastore's {@code timestamp} carries no + * precision at all, so every timestamp column reads back at the precision the connector + * requests from the metastore rather than the one it was declared with.</li> + * </ul> + * Types rejected outright, because the alternative is a mapping that is quietly wrong rather than + * merely wider: + * <ul> + * <li>{@code CHAR(n)} -- Avro has no fixed-width string, so the blank-padding semantics that + * distinguish {@code CHAR} from {@code VARCHAR} would be silently dropped.</li> + * <li>{@code TIMESTAMP(p) WITH TIME ZONE} -- Avro's timestamp logical types carry an instant, not + * an instant plus a zone, so the per-value zone would be lost.</li> + * <li>{@code TIMESTAMP(p)} beyond microsecond precision -- Avro's nanosecond logical types are not + * among those the connector's read path decodes, so such a column would be written and then be + * unreadable.</li> + * <li>{@code UUID} and {@code TIME(p)} -- both have a faithful Avro logical type, but neither has a + * Hive counterpart, so {@code HiveTypeTranslator#toHiveType} rejects them and the column could + * never be registered in the metastore. Since {@code HudiMetadata#getColumnHandles} reads + * columns from the metastore, such a column would also be unreadable. Rejecting here keeps the + * failure at the column that caused it.</li> + * <li>{@code MAP} with a non-{@code VARCHAR} key type -- Avro map keys are always strings.</li> + * <li>Unnamed {@code ROW} fields -- Avro record fields must be named.</li> + * </ul> + * {@code DECIMAL} maps to Avro {@code bytes} rather than {@code fixed}. Both are decodable by the + * read path, and {@code bytes} is what Hudi's own {@link HoodieSchema#createDecimal(int, int)} + * helper produces; {@code fixed} would additionally require inventing a unique schema name per + * decimal column, since Avro fixed types are named and must not collide within one schema. + * <p> + * Nullability: Trino carries nullability per column but not per element inside a {@code ROW}, + * {@code ARRAY} or {@code MAP}. A column's own {@link ColumnMetadata#isNullable()} is honoured at + * the top level; everything nested is made nullable, which is what Spark's Avro conversion also + * produces. + */ +public final class HudiSchemaConverter +{ + private static final String NAMESPACE = "hoodie.trino"; + private static final int MAX_MILLIS_PRECISION = 3; + private static final int MAX_MICROS_PRECISION = 6; + + private HudiSchemaConverter() {} + + /** + * Builds the table schema, with Hudi's five meta fields prepended exactly as + * {@link HoodieSchemaUtils#addMetadataFields} would for any other engine. + * <p> + * The returned schema is the single source of truth for both {@code hoodie.table.create.schema} + * and the Hive Metastore column list. Deriving those two from separate inputs is what produces a + * table whose metastore descriptor and Hudi schema disagree (HUDI-9435). + * + * @param columns all table columns in declaration order, partition columns included + * @param tableName used to name the Avro record + */ + public static HoodieSchema toTableSchema(List<ColumnMetadata> columns, String tableName) + { + String recordName = sanitizeName(tableName); + RecordNameAllocator recordNames = new RecordNameAllocator(recordName); + ImmutableList.Builder<HoodieSchemaField> fields = ImmutableList.builder(); + for (ColumnMetadata column : columns) { + HoodieSchema fieldSchema = toHoodieSchema(column.getType(), column.getName(), recordNames); + if (column.isNullable()) { + fields.add(HoodieSchemaField.of( + column.getName(), + HoodieSchema.createNullable(fieldSchema), + column.getComment().orElse(null), + HoodieSchema.NULL_VALUE)); + } + else { + fields.add(HoodieSchemaField.of(column.getName(), fieldSchema, column.getComment().orElse(null), null)); Review Comment: **minor:** Column and ROW field names go straight into `new Schema.Field`, so Trino-legal quoted identifiers like `"a-b"`, `"a b"` or `"1a"` fail with a raw Avro `SchemaParseException` ("Illegal character") that surfaces as an internal error. The `testColumnName` override skips the CREATE half of the contract test, which feeds exactly these names. Not blocking, but could `HudiTableValidation` reject invalid Avro names with `TrinoException(NOT_SUPPORTED)` naming the column, with `isColumnNameRejected` overridden to match? ########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiMetastoreTables.java: ########## @@ -0,0 +1,204 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hudi; + +import com.google.common.collect.ImmutableList; +import io.trino.metastore.Column; +import io.trino.metastore.Table; +import io.trino.plugin.hudi.util.HudiSchemaConverter; +import io.trino.plugin.hudi.util.HudiTableTypeUtils; +import io.trino.spi.TrinoException; +import io.trino.spi.connector.ColumnMetadata; +import io.trino.spi.connector.SchemaTableName; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.schema.HoodieSchemaField; +import org.apache.hudi.sync.common.util.SparkDataSourceTableUtils; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import static io.trino.metastore.HiveType.HIVE_LONG; +import static io.trino.metastore.HiveType.HIVE_STRING; +import static io.trino.plugin.hive.TableType.EXTERNAL_TABLE; +import static io.trino.spi.type.BigintType.BIGINT; +import static io.trino.spi.type.TimestampType.createTimestampType; +import static io.trino.spi.type.VarcharType.VARCHAR; +import static org.apache.hudi.common.model.HoodieRecord.COMMIT_TIME_METADATA_FIELD; +import static org.apache.hudi.common.model.HoodieRecord.RECORD_KEY_METADATA_FIELD; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Guards the invariant this class exists for: the metastore descriptor and + * {@code hoodie.table.create.schema} describe the same columns. {@code getColumnHandles} reads only + * the metastore, so a mismatch is invisible until a query returns wrong columns -- or none, as in + * HUDI-9435. + */ +final class TestHudiMetastoreTables +{ + private static final SchemaTableName TABLE_NAME = new SchemaTableName("sales", "trips"); + private static final String BASE_PATH = "memory:///warehouse/trips"; + + @Test + void testEveryHudiSchemaFieldIsRegistered() + { + // The HUDI-9435 guard. Data columns plus partition columns must account for every field in + // the Hudi schema, with nothing invented and nothing dropped. + HoodieSchema schema = schema(); + Table table = buildTable(HoodieTableType.COPY_ON_WRITE, ImmutableList.of("city"), schema); + + List<String> registered = ImmutableList.<Column>builder() + .addAll(table.getDataColumns()) + .addAll(table.getPartitionColumns()) + .build().stream() + .map(Column::getName) + .toList(); + assertThat(registered) + .containsExactlyInAnyOrderElementsOf(schema.getFields().stream().map(HoodieSchemaField::name).toList()); + } + + @Test + void testMetaFieldsAreRegisteredAndLeadTheDataColumns() + { + Table table = buildTable(HoodieTableType.COPY_ON_WRITE, ImmutableList.of(), schema()); + + List<String> dataColumns = table.getDataColumns().stream().map(Column::getName).toList(); + assertThat(dataColumns).startsWith(COMMIT_TIME_METADATA_FIELD); + assertThat(dataColumns).contains(RECORD_KEY_METADATA_FIELD); + // Registered as strings, matching what hive sync produces for Spark- and Flink-created tables. + assertThat(columnType(table, COMMIT_TIME_METADATA_FIELD)).isEqualTo(HIVE_STRING); + } + + @Test + void testPartitionColumnsAreSplitOutInDeclaredOrder() + { + Table table = buildTable(HoodieTableType.COPY_ON_WRITE, ImmutableList.of("city", "id"), schema()); + + assertThat(table.getPartitionColumns().stream().map(Column::getName).toList()) + .containsExactly("city", "id"); + assertThat(table.getDataColumns().stream().map(Column::getName).toList()) + .doesNotContain("city", "id"); + } + + @Test + void testColumnTypesComeFromTheSchema() + { + Table table = buildTable(HoodieTableType.COPY_ON_WRITE, ImmutableList.of(), schema()); + + assertThat(columnType(table, "id")).isEqualTo(HIVE_LONG); + assertThat(columnType(table, "city")).isEqualTo(HIVE_STRING); + } + + @ParameterizedTest + @EnumSource(HoodieTableType.class) + void testInputFormatRoundTripsToTheTableType(HoodieTableType tableType) + { + // The input format is the only record of the table type in the metastore, and the only signal + // isHudiTable uses. If this did not round-trip, getTableHandle would reject the table the + // connector had just created. + Table table = buildTable(tableType, ImmutableList.of(), schema()); + + String inputFormat = table.getStorage().getStorageFormat().getInputFormat(); + assertThat(HudiTableTypeUtils.fromInputFormat(inputFormat)).isEqualTo(tableType); + } + + @Test + void testRegisteredAsExternalTable() + { + Table table = buildTable(HoodieTableType.COPY_ON_WRITE, ImmutableList.of(), schema()); + + assertThat(table.getTableType()).isEqualTo(EXTERNAL_TABLE.name()); + assertThat(table.getParameters()).containsEntry("EXTERNAL", "TRUE"); + assertThat(table.getStorage().getLocation()).isEqualTo(BASE_PATH); + } + + @Test + void testPartitionColumnMissingFromSchemaIsRegisteredAsString() + { + Table table = buildTable(HoodieTableType.COPY_ON_WRITE, ImmutableList.of("region"), schema()); + + assertThat(table.getPartitionColumns().stream().map(Column::getName).toList()) + .containsExactly("region"); + assertThat(columnType(table, "region")).isEqualTo(HIVE_STRING); + } + + @Test + void testRepeatedPartitionColumnIsRejected() + { + assertThatThrownBy(() -> buildTable(HoodieTableType.COPY_ON_WRITE, ImmutableList.of("city", "city"), schema())) + .isInstanceOf(TrinoException.class) + .hasMessageContaining("listed more than once"); + } + + @Test + void testSharedSparkDataSourcePropertiesAreUsableFromTheConnector() Review Comment: **minor:** This calls `SparkDataSourceTableUtils` directly and never goes through `HudiMetastoreTables`, so it re-asserts `TestHoodieMetastoreTableDescriptor` (lines 166-179) and would not catch broken connector wiring. Not blocking, but could it instead assert that `HudiMetastoreTables.buildTable(...).getParameters()` contains `spark.sql.sources.provider=hudi`? ########## hudi-trino/src/main/java/io/trino/plugin/hudi/HudiTableHandle.java: ########## @@ -116,9 +116,12 @@ public HudiTableHandle( .filterCompletedInstants() .lastInstant() .map(HoodieInstant::requestedTime) - .orElseThrow(() -> new TrinoException( - HudiErrorCode.HUDI_NO_VALID_COMMIT, - "Table has no valid commits"))); + // An initialized table has a schema but intentionally has no data commit. + // Hudi uses INIT_INSTANT_TS as the lower bound for this state. Passing it to + // the file-system view produces no file slices, so empty native CREATE TABLE + // and newly registered empty tables remain queryable without fabricating a + // commit or entering the row-writing path. + .orElse(INIT_INSTANT_TS)); Review Comment: **nit:** With `orElse(INIT_INSTANT_TS)`, a table whose only commit is inflight or rolled back now reads as empty instead of failing, and `HudiErrorCode.HUDI_NO_VALID_COMMIT` is left unused. Feel free to ignore, but could we remove the error code, or add a test pinning the inflight-only case? ########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiDdl.java: ########## @@ -0,0 +1,277 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hudi; + +import io.trino.filesystem.Location; +import io.trino.filesystem.TrinoFileSystem; +import io.trino.filesystem.TrinoFileSystemFactory; +import io.trino.metastore.HiveMetastore; +import io.trino.metastore.HiveMetastoreFactory; +import io.trino.plugin.hudi.storage.TrinoStorageConfiguration; +import io.trino.plugin.hudi.testing.HudiTablesInitializer; +import io.trino.spi.security.ConnectorIdentity; +import io.trino.testing.AbstractTestQueryFramework; +import io.trino.testing.QueryRunner; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.schema.HoodieSchemaField; +import org.apache.hudi.common.schema.HoodieSchemaType; +import org.apache.hudi.common.schema.HoodieSchemaUtils; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.HoodieTableVersion; +import org.apache.hudi.storage.StoragePath; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.Optional; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.apache.hudi.common.model.HoodieTableType.COPY_ON_WRITE; +import static org.apache.hudi.common.model.HoodieTableType.MERGE_ON_READ; + +final class TestHudiDdl + extends AbstractTestQueryFramework +{ + private final UnregisteredTableInitializer initializer = new UnregisteredTableInitializer(); + + @Override + protected QueryRunner createQueryRunner() + throws Exception + { + return HudiQueryRunner.builder() + .setDataLoader(initializer) + .build(); + } + + @Test + void testRegisterAndUnregisterExistingTableWithoutChangingStorage() + { + String tableName = initializer.getTableName(); + String tableLocation = initializer.getTableLocation().toString(); + + assertUpdate("CALL hudi.system.register_table(" + + "schema_name => 'tests', " + + "table_name => '" + tableName + "', " + + "table_location => '" + tableLocation + "')"); + assertQueryFails( + "CALL hudi.system.register_table('tests', '" + tableName + "', '" + tableLocation + "')", + ".*Table already exists: tests\\." + tableName + ".*"); + + assertQuery("SELECT count(*) FROM " + tableName, "VALUES CAST(0 AS BIGINT)"); + assertThat(initializer.getMetastore().getTable("tests", tableName)).get() + .satisfies(table -> { + assertThat(table.getStorage().getLocation()).isEqualTo(tableLocation); + assertThat(table.getPartitionColumns()).extracting(io.trino.metastore.Column::getName) + .containsExactly("city"); + assertThat(table.getParameters()).containsEntry("EXTERNAL", "TRUE"); + }); + assertThat(initializer.loadMetaClient().getTableConfig().getTableVersion()) + .isEqualTo(HoodieTableVersion.EIGHT); + + assertUpdate("CALL hudi.system.unregister_table(" + + "schema_name => 'tests', " + + "table_name => '" + tableName + "')"); + + assertThat(initializer.getMetastore().getTable("tests", tableName)).isEmpty(); + assertThat(HudiUtil.hudiMetadataExists(initializer.getFileSystem(), initializer.getTableLocation())).isTrue(); + assertThat(initializer.loadMetaClient().getTableConfig().getTableVersion()) + .isEqualTo(HoodieTableVersion.EIGHT); + assertQueryFails( + "CALL hudi.system.unregister_table('tests', '" + tableName + "')", + ".*Table 'tests\\." + tableName + "' not found.*"); + + assertUpdate("CALL hudi.system.register_table('tests', '" + tableName + "', '" + tableLocation + "')"); + assertUpdate("DROP TABLE " + tableName); + + assertThat(initializer.getMetastore().getTable("tests", tableName)).isEmpty(); + assertThat(HudiUtil.hudiMetadataExists(initializer.getFileSystem(), initializer.getTableLocation())).isTrue(); + } + + @Test + void testCreateAndDropExternalMergeOnReadTablePreservesStorage() + { + String tableName = "created_external_mor"; + Location tableLocation = initializer.getExternalLocation().appendPath(tableName); + + assertUpdate(""" + CREATE TABLE %s ( + id bigint, + name varchar, + city varchar + ) + WITH ( + location = '%s', + table_type = 'MERGE_ON_READ', + partitioned_by = ARRAY['city'], + primary_key = ARRAY['id'], + ordering_fields = ARRAY['id'] + ) + """.formatted(tableName, tableLocation)); + + assertQuery("SELECT count(*) FROM " + tableName, "VALUES CAST(0 AS BIGINT)"); + assertThat(initializer.getMetastore().getTable("tests", tableName)).get() + .satisfies(table -> { + assertThat(table.getTableType()).isEqualTo("EXTERNAL_TABLE"); + assertThat(table.getStorage().getLocation()).isEqualTo(tableLocation.toString()); + assertThat(table.getPartitionColumns()).extracting(io.trino.metastore.Column::getName) + .containsExactly("city"); + }); + assertThat(initializer.loadMetaClient(tableName, tableLocation).getTableConfig()) + .satisfies(config -> { + assertThat(config.getTableType()).isEqualTo(MERGE_ON_READ); + assertThat(config.getTableVersion()).isEqualTo(HudiTableInitializer.CREATED_TABLE_VERSION); + assertThat(config.getPartitionFields().get()).containsExactly("city"); + assertThat(config.getRecordKeyFields().get()).containsExactly("id"); + assertThat(config.getOrderingFields()).containsExactly("id"); + }); + + assertUpdate("DROP TABLE " + tableName); + + assertThat(initializer.getMetastore().getTable("tests", tableName)).isEmpty(); + assertThat(HudiUtil.hudiMetadataExists(initializer.getFileSystem(), tableLocation)).isTrue(); + } + + @Test + void testCreateAndDropManagedCopyOnWriteTableDeletesStorage() + { + String tableName = "created_managed_cow"; + String schemaLocation = initializer.getMetastore().getDatabase("tests").orElseThrow().getLocation().orElseThrow(); + Location tableLocation = Location.of(schemaLocation).appendPath(tableName); + + assertUpdate("CREATE TABLE " + tableName + " (id bigint, name varchar)"); + + assertQuery("SELECT count(*) FROM " + tableName, "VALUES CAST(0 AS BIGINT)"); + assertThat(initializer.getMetastore().getTable("tests", tableName)).get() + .satisfies(table -> { + assertThat(table.getTableType()).isEqualTo("MANAGED_TABLE"); + assertThat(table.getParameters()).doesNotContainKey("EXTERNAL"); + assertThat(table.getStorage().getLocation()).isEqualTo(tableLocation.toString()); + }); + assertThat(initializer.loadMetaClient(tableName, tableLocation).getTableConfig()) + .satisfies(config -> { + assertThat(config.getTableType()).isEqualTo(COPY_ON_WRITE); + assertThat(config.getTableVersion()).isEqualTo(HudiTableInitializer.CREATED_TABLE_VERSION); + }); + + assertQueryFails( Review Comment: **nit:** Some duplication, feel free to ignore: - These INSERT/CTAS rejection asserts repeat contract tests that already run with those flags off (`BaseConnectorSmokeTest`, `BaseConnectorTest`). - Four `TestHudiMetastoreTables` tests (meta fields, partition order, missing partition column, repeated partition) re-assert `TestHoodieMetastoreTableDescriptor` through the wrapper; ordering-vs-precombine is checked in both `TestHudiTableValidation` and `TestHudiTableProperties`. - The COW/MOR pairs and the five reject-type tests in `TestHudiSchemaConverter` could be `@EnumSource` / `@MethodSource`. - The injector metastore lookup at line 213 is a 6th copy (see the `*HudiTablesInitializer` classes), and fixed table names without try/finally leak on failure. Could we trim these? ########## hudi-trino/src/main/java/io/trino/plugin/hudi/procedure/RegisterTableProcedure.java: ########## @@ -0,0 +1,147 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hudi.procedure; + +import com.google.common.collect.ImmutableList; +import com.google.inject.Inject; +import com.google.inject.Provider; +import io.trino.filesystem.Location; +import io.trino.filesystem.TrinoFileSystem; +import io.trino.filesystem.TrinoFileSystemFactory; +import io.trino.metastore.HiveMetastore; +import io.trino.metastore.HiveMetastoreFactory; +import io.trino.plugin.hudi.HudiMetastoreTables; +import io.trino.plugin.hudi.HudiUtil; +import io.trino.spi.TrinoException; +import io.trino.spi.classloader.ThreadContextClassLoader; +import io.trino.spi.connector.ConnectorAccessControl; +import io.trino.spi.connector.ConnectorSession; +import io.trino.spi.connector.SchemaNotFoundException; +import io.trino.spi.connector.SchemaTableName; +import io.trino.spi.procedure.Procedure; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.table.HoodieTableMetaClient; + +import java.lang.invoke.MethodHandle; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import static io.trino.metastore.PrincipalPrivileges.NO_PRIVILEGES; +import static io.trino.plugin.base.util.Procedures.checkProcedureArgument; +import static io.trino.plugin.hudi.HudiTableProperties.LOCATION_PROPERTY; +import static io.trino.spi.StandardErrorCode.ALREADY_EXISTS; +import static io.trino.spi.type.VarcharType.VARCHAR; +import static java.lang.invoke.MethodHandles.lookup; +import static java.util.Objects.requireNonNull; + +public class RegisterTableProcedure + implements Provider<Procedure> +{ + private static final MethodHandle REGISTER_TABLE; + + static { + try { + REGISTER_TABLE = lookup().unreflect(RegisterTableProcedure.class.getMethod( + "registerTable", + ConnectorSession.class, + ConnectorAccessControl.class, + String.class, + String.class, + String.class)); + } + catch (ReflectiveOperationException e) { + throw new AssertionError(e); + } + } + + private final HiveMetastoreFactory metastoreFactory; + private final TrinoFileSystemFactory fileSystemFactory; + + @Inject + public RegisterTableProcedure(HiveMetastoreFactory metastoreFactory, TrinoFileSystemFactory fileSystemFactory) + { + this.metastoreFactory = requireNonNull(metastoreFactory, "metastoreFactory is null"); + this.fileSystemFactory = requireNonNull(fileSystemFactory, "fileSystemFactory is null"); + } + + @Override + public Procedure get() + { + return new Procedure( + "system", + "register_table", + ImmutableList.of( + new Procedure.Argument("SCHEMA_NAME", VARCHAR), + new Procedure.Argument("TABLE_NAME", VARCHAR), + new Procedure.Argument("TABLE_LOCATION", VARCHAR)), + REGISTER_TABLE.bindTo(this)); + } + + public void registerTable( + ConnectorSession session, + ConnectorAccessControl accessControl, + String schemaName, + String tableName, + String tableLocation) + { + try (ThreadContextClassLoader _ = new ThreadContextClassLoader(getClass().getClassLoader())) { + doRegisterTable(session, accessControl, schemaName, tableName, tableLocation); + } + } + + private void doRegisterTable( + ConnectorSession session, + ConnectorAccessControl accessControl, + String schemaName, + String tableName, + String tableLocation) + { + checkProcedureArgument(schemaName != null, "schema_name cannot be null"); Review Comment: **major:** `register_table` is always on, so any user with CREATE TABLE on a schema can expose any path the connector's credentials can read. Trino's Delta and Iceberg connectors gate the same procedure behind `delta.register-table-procedure.enabled` / `iceberg.register-table-procedure.enabled`, both default `false`. Could we add a `hudi.register-table-procedure.enabled` config (default `false`) and check it first here, as Delta's `RegisterTableProcedure` does? ########## hudi-trino/src/main/java/io/trino/plugin/hudi/HudiTableInitializer.java: ########## @@ -0,0 +1,116 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hudi; + +import io.trino.filesystem.TrinoFileSystem; +import io.trino.plugin.hudi.storage.TrinoStorageConfiguration; +import io.trino.spi.TrinoException; +import io.trino.spi.connector.ConnectorTableMetadata; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.HoodieTableVersion; +import org.apache.hudi.storage.StoragePath; + +import java.io.IOException; +import java.util.List; +import java.util.Map; + +import static io.trino.plugin.hudi.HudiErrorCode.HUDI_META_CLIENT_ERROR; +import static io.trino.plugin.hudi.HudiTableProperties.getHiveStylePartitioning; +import static io.trino.plugin.hudi.HudiTableProperties.getHoodieProperties; +import static io.trino.plugin.hudi.HudiTableProperties.getKeyGeneratorClass; +import static io.trino.plugin.hudi.HudiTableProperties.getOrderingFields; +import static io.trino.plugin.hudi.HudiTableProperties.getPartitionedBy; +import static io.trino.plugin.hudi.HudiTableProperties.getPrimaryKey; +import static io.trino.plugin.hudi.HudiTableProperties.getRecordMergeMode; +import static io.trino.plugin.hudi.HudiTableProperties.getTableType; +import static java.lang.String.format; + +/** + * Writes a table's {@code .hoodie} directory. Pure hudi-common: no Spark, no write client and no + * metastore, so it can run before the catalog entry exists without entering any row-writing path. + */ +public final class HudiTableInitializer +{ + /** + * The {@code hoodie.table.version} written by this connector. + * <p> + * Pinned rather than inherited from {@link HoodieTableVersion#current()} so that a future bump of + * {@code current()} cannot silently change what the connector emits. Raising this should be a + * deliberate change with a test diff, not a side effect of upgrading Hudi; the assertion in + * {@code TestHudiDdl} fails when the two drift. + */ + public static final HoodieTableVersion CREATED_TABLE_VERSION = HoodieTableVersion.TEN; + + private HudiTableInitializer() {} + + /** + * Initializes storage for an empty table. + * + * @param tableSchema the schema that also produces the metastore column list, so the two cannot + * disagree + */ + public static void initializeTable( + TrinoFileSystem fileSystem, + String basePath, + ConnectorTableMetadata tableMetadata, + HoodieSchema tableSchema) + { + Map<String, Object> properties = tableMetadata.getProperties(); + HoodieTableMetaClient.TableBuilder builder = HoodieTableMetaClient.newTableBuilder() + .setTableType(getTableType(properties)) + .setTableName(tableMetadata.getTable().getTableName()) + .setDatabaseName(tableMetadata.getTable().getSchemaName()) + .setTableVersion(CREATED_TABLE_VERSION) + .setTableCreateSchema(tableSchema.toAvroSchema().toString()); Review Comment: **major:** `tableSchema` already carries the five `_hoodie_*` fields (`toTableSchema` ends with `addMetadataFields`), but since HUDI-6073 (#8450) `hoodie.table.create.schema` holds data columns only; Spark strips them in `HoodieCatalogTable.initHoodieTable`. Until the first commit, `TableSchemaResolver.getTableSchema(false)` returns this schema as-is, so the Flink catalog, `SecondaryIndexManager` and incremental relations see the meta fields as data columns. Could we pass `HoodieSchemaUtils.removeMetadataFields(tableSchema)` here and flip `testCreateSchemaCarriesMetaFieldsAndDataColumns` to `doesNotContain`? ########## hudi-trino/src/main/java/io/trino/plugin/hudi/HudiMetastoreTables.java: ########## @@ -0,0 +1,133 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hudi; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import io.trino.metastore.Column; +import io.trino.metastore.StorageFormat; +import io.trino.metastore.Table; +import io.trino.spi.TrinoException; +import io.trino.spi.connector.SchemaTableName; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.schema.HoodieSchemaField; +import org.apache.hudi.sync.common.util.HoodieMetastoreTableDescriptor; + +import java.util.List; +import java.util.Optional; + +import static io.trino.plugin.hive.TableType.EXTERNAL_TABLE; +import static io.trino.plugin.hive.TableType.MANAGED_TABLE; +import static io.trino.plugin.hudi.HudiUtil.toColumnHandle; +import static io.trino.spi.StandardErrorCode.INVALID_TABLE_PROPERTY; + +/** + * Translates {@link HoodieMetastoreTableDescriptor} into a Trino {@link Table}. + * <p> + * The assembly decisions -- which input format carries the table type, which columns are partition + * columns, which properties make the table recognisable to Spark SQL, where + * {@code serialization.format} goes -- are not made here. They live in {@code hudi-sync-common} + * next to {@code SparkDataSourceTableUtils}, in Hudi and JDK types, so that the normalization contract + * is reusable by metastore integrations without depending on Trino types. All that happens here is + * the mapping into {@code io.trino.metastore} types. Parity tests in {@code hudi-hive-sync} pin the + * shared format names and schema-splitting default to the values its existing executor emits. + * <p> + * The columns come from the same {@link HoodieSchema} that becomes + * {@code hoodie.table.create.schema}, which is the one thing this class does insist on. The + * connector keeps two independent descriptions of a table's columns -- that schema, and the + * metastore storage descriptor -- and {@link HudiMetadata#getColumnHandles} reads only the latter. + * Nothing reconciles them, so a table built from two separately-handled inputs can end up registered + * with columns that do not match its own schema, or with none at all (HUDI-9435: + * {@code SELECT * not allowed from relation that has no columns}). Both {@code CREATE TABLE} and + * {@code register_table} go through here so that cannot happen. + * <p> + * Column types come from {@link HudiUtil#toColumnHandle}, the same Avro-to-Trino mapping the read + * path uses, so the types registered are the types a query will see. + */ +public final class HudiMetastoreTables +{ + private HudiMetastoreTables() {} + + /** + * Builds the metastore descriptor for a table's snapshot view. + * <p> + * Lenient about a partition column that is absent from {@code tableSchema}, typing it as a + * string, because the shared layer is and because {@code register_table} has to cope with real + * tables built by key generators that do not record their partition fields in the schema. + * {@code CREATE TABLE} rejects that case earlier, in {@link HudiTableValidation}, where the + * offending property can be named. + * + * @param tableSchema the table's Hudi schema, including the meta fields; supplies both the + * column names and their types + * @param partitionedBy partition column names, in partition-path order + * @param external whether the table is external. An explicit {@code location} makes a table + * external, an omitted one makes it managed; see {@link HudiMetadata#createTable}. + */ + public static Table buildTable( + SchemaTableName tableName, + String basePath, + HoodieTableType tableType, + HoodieSchema tableSchema, + List<String> partitionedBy, + boolean external, + Optional<String> owner, + Optional<String> comment) + { + HoodieMetastoreTableDescriptor descriptor; + try { + descriptor = HoodieMetastoreTableDescriptor.forSnapshotView( + tableSchema, partitionedBy, tableType, basePath, external); + } + catch (IllegalArgumentException e) { + throw new TrinoException(INVALID_TABLE_PROPERTY, e.getMessage(), e); + } + + ImmutableMap.Builder<String, String> parameters = ImmutableMap.<String, String>builder() + .putAll(descriptor.getTableParameters()); + comment.ifPresent(value -> parameters.put(Table.TABLE_COMMENT, value)); + + return Table.builder() + .setDatabaseName(tableName.getSchemaName()) + .setTableName(tableName.getTableName()) + // Hive treats the table type and the EXTERNAL parameter as independent; the shared + // layer supplies the parameter, this supplies the type, and they must agree. + .setTableType((external ? EXTERNAL_TABLE : MANAGED_TABLE).name()) + .setOwner(owner) + .setDataColumns(toColumns(descriptor.getDataFields())) + .setPartitionColumns(toColumns(descriptor.getPartitionFields())) + .setParameters(parameters.buildKeepingLast()) + .withStorage(storage -> storage + .setStorageFormat(StorageFormat.create( + descriptor.getSerdeClassName(), + descriptor.getInputFormatClassName(), + descriptor.getOutputFormatClassName())) + .setSerdeParameters(descriptor.getSerdeParameters()) + .setLocation(basePath)) + .build(); + } + + private static List<Column> toColumns(List<HoodieSchemaField> fields) + { + ImmutableList.Builder<Column> columns = ImmutableList.builderWithExpectedSize(fields.size()); + for (HoodieSchemaField field : fields) { + columns.add(new Column( + field.name(), + toColumnHandle(field).getHiveType(), Review Comment: **minor:** For `register_table`, the Hive types come from Trino's Avro mapping rather than hive-sync's `HiveSchemaUtil`, and the two seem to disagree: `local-timestamp-*` falls back to `bigint` in Trino (hive-sync writes `TIMESTAMP`), and TIME/UUID make `HiveTypeTranslator` throw. The only register test uses bigint/varchar on an empty table. Not blocking, but could we add a `register_table` test on a table with commits and timestamp/decimal/row columns, and reject or document the divergent types? ########## hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/util/HoodieMetastoreTableDescriptor.java: ########## @@ -0,0 +1,399 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hudi.sync.common.util; + +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.schema.HoodieSchemaField; +import org.apache.hudi.common.schema.HoodieSchemaType; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * The engine-neutral description of how a Hudi table is registered in a Hive-style metastore. + * + * <p>Registering a Hudi table means agreeing on a handful of things that have nothing to do with any + * particular engine: which input format carries the table type, which columns are partition columns + * and which are data columns, and which properties make the table recognisable to Spark SQL. Those + * decisions were previously made independently by {@code HMSDDLExecutor} (for hive-sync), by Spark + * SQL's {@code createHiveDataSourceTable}, and by Flink's {@code TableOptionProperties}. Each + * arrived at nearly the same answer, and "nearly" is the problem: a table that disagrees on any one + * of them is a table another engine either cannot read or does not recognise as Hudi at all. + * + * <p>This class exposes a normalized form of those decisions in Hudi and JDK types only. It + * deliberately carries no Hive types: the metastore column type is a {@link HoodieSchemaField} here, + * and the input/output format and serde are class <em>names</em> rather than classes, so a caller that has no + * {@code hive-metastore} on its classpath -- the Trino connector, for one -- can consume it and + * translate to its own metastore types. {@code hudi-hadoop-mr} holds the only existing copy of the + * format-name mapping ({@code HoodieInputFormatUtils}), and it is not reachable from such a caller, + * so the names are repeated here as constants and pinned by + * {@code TestHoodieMetastoreTableDescriptorFormatNames} against the values hive-sync emits. + * + * <p>The Spark datasource properties come from {@link SparkDataSourceTableUtils} rather than being + * rebuilt, since that class is already engine-neutral and its Spark compatibility is already + * covered by round-tripping the schema JSON through Spark's own {@code StructType.fromJson}. + * + * <p>Only the Parquet base file format is described. Every engine that creates a table today creates + * a Parquet one, and reproducing the ORC/HFile/Lance/Vortex branches of + * {@code HoodieInputFormatUtils} without a caller for them would be speculative; a non-Parquet base Review Comment: **minor:** The javadoc says a non-Parquet base file format is rejected, but nothing checks it: `register_table` on an ORC/HFile/Lance base-format table writes `HoodieParquetInputFormat` and `ParquetHiveSerDe` to HMS. Spark's `CreateHoodieTableCommand` derives the formats from `getBaseFileFormat`. Not blocking, but could `register_table` reject `metaClient.getTableConfig().getBaseFileFormat() != PARQUET` with a `TrinoException`, plus a test? ########## hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/util/HoodieMetastoreTableDescriptor.java: ########## @@ -0,0 +1,399 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hudi.sync.common.util; + +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.schema.HoodieSchemaField; +import org.apache.hudi.common.schema.HoodieSchemaType; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * The engine-neutral description of how a Hudi table is registered in a Hive-style metastore. + * + * <p>Registering a Hudi table means agreeing on a handful of things that have nothing to do with any + * particular engine: which input format carries the table type, which columns are partition columns + * and which are data columns, and which properties make the table recognisable to Spark SQL. Those + * decisions were previously made independently by {@code HMSDDLExecutor} (for hive-sync), by Spark + * SQL's {@code createHiveDataSourceTable}, and by Flink's {@code TableOptionProperties}. Each + * arrived at nearly the same answer, and "nearly" is the problem: a table that disagrees on any one + * of them is a table another engine either cannot read or does not recognise as Hudi at all. + * + * <p>This class exposes a normalized form of those decisions in Hudi and JDK types only. It + * deliberately carries no Hive types: the metastore column type is a {@link HoodieSchemaField} here, + * and the input/output format and serde are class <em>names</em> rather than classes, so a caller that has no + * {@code hive-metastore} on its classpath -- the Trino connector, for one -- can consume it and + * translate to its own metastore types. {@code hudi-hadoop-mr} holds the only existing copy of the + * format-name mapping ({@code HoodieInputFormatUtils}), and it is not reachable from such a caller, + * so the names are repeated here as constants and pinned by + * {@code TestHoodieMetastoreTableDescriptorFormatNames} against the values hive-sync emits. + * + * <p>The Spark datasource properties come from {@link SparkDataSourceTableUtils} rather than being + * rebuilt, since that class is already engine-neutral and its Spark compatibility is already + * covered by round-tripping the schema JSON through Spark's own {@code StructType.fromJson}. + * + * <p>Only the Parquet base file format is described. Every engine that creates a table today creates + * a Parquet one, and reproducing the ORC/HFile/Lance/Vortex branches of + * {@code HoodieInputFormatUtils} without a caller for them would be speculative; a non-Parquet base + * file format is rejected rather than guessed at. + */ +public final class HoodieMetastoreTableDescriptor { + + /** + * Input format for a Copy-on-Write table, and for the read-optimized view of a Merge-on-Read one. + * + * <p>The input format is the only record of the table type in the metastore, so these four names + * are load-bearing well beyond format selection. Trino's {@code HiveUtil#isHudiTable} recognises + * a Hudi table by this field alone. + */ + public static final String PARQUET_INPUT_FORMAT_CLASS = "org.apache.hudi.hadoop.HoodieParquetInputFormat"; + + /** Input format for the real-time (snapshot) view of a Merge-on-Read table. */ + public static final String PARQUET_REALTIME_INPUT_FORMAT_CLASS = "org.apache.hudi.hadoop.realtime.HoodieParquetRealtimeInputFormat"; + + public static final String PARQUET_OUTPUT_FORMAT_CLASS = "org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat"; + + public static final String PARQUET_SERDE_CLASS = "org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe"; + + /** + * Marks the table as external. Set alongside the metastore's own table type, not instead of it: + * Hive treats the two as independent, and a table with {@code EXTERNAL_TABLE} but no + * {@code EXTERNAL=TRUE} parameter is still dropped destructively by some metastore versions. + */ + public static final String EXTERNAL_PARAMETER = "EXTERNAL"; + + public static final String EXTERNAL_PARAMETER_VALUE = "TRUE"; + + /** + * Written into the <em>serde</em> parameters, which is where {@code HMSDDLExecutor} puts it. It is + * a serde-level setting, and a table carrying it in its table parameters instead differs from + * every hive-synced table without failing in any way that is visible until an engine compares the + * two. + */ + public static final String SERIALIZATION_FORMAT_PARAMETER = "serialization.format"; + + public static final String SERIALIZATION_FORMAT_VALUE = "1"; + + /** + * Matches {@code hoodie.datasource.hive_sync.schema_string_length_thresh}, the chunk size the + * Spark-reconstructible schema is split into across {@code spark.sql.sources.schema.part.N}. + * Spark reassembles by concatenating the parts in order, so the value only has to be stable, not + * any particular number -- but keeping it equal to the hive-sync default keeps a Trino-created + * table byte-identical to a hive-synced one. + */ + public static final int DEFAULT_SCHEMA_STRING_LENGTH_THRESHOLD = 4000; + + private final List<HoodieSchemaField> dataFields; + private final List<HoodieSchemaField> partitionFields; + private final String inputFormatClassName; + private final String outputFormatClassName; + private final String serdeClassName; + private final Map<String, String> tableParameters; + private final Map<String, String> serdeParameters; + private final boolean external; + + private HoodieMetastoreTableDescriptor( + List<HoodieSchemaField> dataFields, + List<HoodieSchemaField> partitionFields, + String inputFormatClassName, + String outputFormatClassName, + String serdeClassName, + Map<String, String> tableParameters, + Map<String, String> serdeParameters, + boolean external) { + this.dataFields = Collections.unmodifiableList(dataFields); + this.partitionFields = Collections.unmodifiableList(partitionFields); + this.inputFormatClassName = inputFormatClassName; + this.outputFormatClassName = outputFormatClassName; + this.serdeClassName = serdeClassName; + this.tableParameters = Collections.unmodifiableMap(tableParameters); + this.serdeParameters = Collections.unmodifiableMap(serdeParameters); + this.external = external; + } + + /** + * Describes the snapshot view of a table: the whole table for Copy-on-Write, the real-time view + * for Merge-on-Read. This is what an engine registering a single table wants; the separate + * read-optimized ({@code _ro}) registration hive-sync also performs for Merge-on-Read is the only + * case that needs {@code readAsOptimized}, so it is not exposed here. + * + * @param tableSchema the table schema <em>including</em> Hudi's meta fields, as + * {@link org.apache.hudi.common.schema.HoodieSchemaUtils#addMetadataFields} produces. Supplies + * both the column names and their types, so a caller cannot register columns that disagree + * with the schema it wrote to {@code hoodie.properties}. + * @param partitionFieldNames partition columns in partition-path order. A name absent from + * {@code tableSchema} is typed as a string, matching {@code HiveSchemaUtil#getPartitionKeyType}; + * real tables built with some key generators do not carry their partition fields in the + * schema, and rejecting those would make an existing table impossible to register. + * @param basePath the table's base path, recorded as the metastore location and as the serde + * {@code path} + * @param external whether to describe the table as external; see {@link #isExternal()} + */ + public static HoodieMetastoreTableDescriptor forSnapshotView( + HoodieSchema tableSchema, + List<String> partitionFieldNames, + HoodieTableType tableType, + String basePath, + boolean external) { + if (tableType == null) { + throw new IllegalArgumentException("tableType is required"); + } + // The snapshot view of a Merge-on-Read table is its real-time view: the one that merges log + // files over the base files. Registering it with the Copy-on-Write input format instead does not + // fail, it silently reads only the base files. + boolean useRealtimeInputFormat = tableType == HoodieTableType.MERGE_ON_READ; + return forView(tableSchema, partitionFieldNames, tableType, basePath, + ViewOptions.builder() Review Comment: **minor:** `forSnapshotView` never sets `includeFieldDocs`, so Trino column comments reach the HMS columns but not `spark.sql.sources.schema.*`, and Spark `DESCRIBE` shows none. Hive-sync passes the comment flag into the same `getSparkTableProperties` call (#19289). Not blocking, but could we set `setIncludeFieldDocs(true)` here, since `HudiMetastoreTables` always writes the comments to HMS, and add a descriptor test that finds a comment in the reassembled schema JSON? ########## hudi-trino/src/main/java/io/trino/plugin/hudi/HudiMetadata.java: ########## @@ -304,6 +321,202 @@ public Optional<Object> getInfo(ConnectorSession session, ConnectorTableHandle t return Optional.of(new HudiTableInfo(table.getSchemaTableName(), table.getTableType().name(), table.getBasePath())); } + /** + * Creates an empty table: its {@code .hoodie} directory on storage and its catalog entry. + * <p> + * This is the single-call path only. {@code beginCreateTable}/{@code finishCreateTable} and a + * page sink are deliberately absent, so {@code CREATE TABLE AS} and {@code INSERT} still fail as + * unsupported; no rows are written here. + * <p> + * An explicit {@code location} makes the table external, matching Hudi's Spark SQL behaviour. An + * omitted one makes it managed, at {@code <schemaLocation>/<tableName>}, which is what decides + * whether {@code DROP TABLE} later deletes the data. + */ + @Override + public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, SaveMode saveMode) + { + SchemaTableName schemaTableName = tableMetadata.getTable(); + if (saveMode == SaveMode.REPLACE) { + // Replacing a table means deciding what happens to the rows already in it, which is the + // write path this connector does not yet have. + throw new TrinoException(NOT_SUPPORTED, "This connector does not support replacing tables"); + } + Database database = metastore.getDatabase(schemaTableName.getSchemaName()) + .orElseThrow(() -> new SchemaNotFoundException(schemaTableName.getSchemaName())); + if (metastore.getTable(schemaTableName.getSchemaName(), schemaTableName.getTableName()).isPresent()) { + if (saveMode == SaveMode.IGNORE) { + return; + } + throw new TrinoException(ALREADY_EXISTS, "Table already exists: " + schemaTableName); + } + + // Everything that can be rejected is rejected before storage is touched, so a bad statement + // leaves nothing behind. + HudiTableValidation.validateCreateTable(tableMetadata); + + Map<String, Object> properties = tableMetadata.getProperties(); + Optional<String> explicitLocation = getTableLocation(properties); + boolean external = explicitLocation.isPresent(); + String basePath = explicitLocation.orElseGet(() -> defaultTableLocation(database, schemaTableName)); + + TrinoFileSystem fileSystem = fileSystemFactory.create(session); + checkLocationIsEmpty(fileSystem, basePath); + + // One schema object produces both hoodie.table.create.schema and the metastore column list, + // so the two cannot disagree (HUDI-9435). + HoodieSchema tableSchema = HudiSchemaConverter.toTableSchema( + tableMetadata.getColumns(), schemaTableName.getTableName()); + Table table = HudiMetastoreTables.buildTable( + schemaTableName, + basePath, + getTableType(properties), + tableSchema, + getPartitionedBy(properties), + external, + Optional.of(session.getUser()), + tableMetadata.getComment()); + try { + HudiTableInitializer.initializeTable(fileSystem, basePath, tableMetadata, tableSchema); + metastore.createTable(table, NO_PRIVILEGES); + } + catch (TableAlreadyExistsException e) { + // Another CREATE TABLE may have initialized the same managed location and won the + // metastore race. Its catalog entry now owns .hoodie, so deleting it would corrupt the + // live table. A different location still belongs to this failed attempt and is safe to + // clean up. + if (!isTableRegisteredAtLocation(schemaTableName, basePath, e)) { + cleanupTableMetadata(fileSystem, basePath, e); + } + throw e; + } + catch (RuntimeException e) { + // Initialization may have written some or all of .hoodie, but no catalog entry from + // this call references it. Left there it would make a retry fail the emptiness check. + // + // Only .hoodie is removed, never the base path: the base path may have been created by + // someone else, and this connector did not create it. On object storage there is no + // directory to delete in any case -- deleteDirectory removes the objects under the + // prefix, which is exactly the set initTable wrote or may have partially written. + cleanupTableMetadata(fileSystem, basePath, e); + throw e; + } + } + + private boolean isTableRegisteredAtLocation(SchemaTableName tableName, String basePath, RuntimeException failure) + { + try { + return metastore.getTable(tableName.getSchemaName(), tableName.getTableName()) + .map(table -> table.getStorage().getLocation().equals(basePath)) + .orElse(false); + } + catch (RuntimeException lookupFailure) { + // When ownership cannot be established, leave storage intact. An orphan is recoverable; + // deleting metadata that a successful concurrent CREATE references is not. + failure.addSuppressed(lookupFailure); + return true; + } + } + + private static void cleanupTableMetadata(TrinoFileSystem fileSystem, String basePath, RuntimeException failure) + { + try { + fileSystem.deleteDirectory(Location.of(appendPath(basePath, METAFOLDER_NAME))); + } + catch (IOException | RuntimeException cleanupFailure) { + failure.addSuppressed(cleanupFailure); + } + } + + /** + * Drops the catalog entry, and the data too when the table is managed. + * <p> + * A table registered with an explicit location is external and its data outlives the catalog + * entry; {@code register_table} always produces such a table. Trino's {@code DROP TABLE} has no + * {@code PURGE} clause, so there is no way to ask for an external table's data to be deleted. + */ + @Override + public void dropTable(ConnectorSession session, ConnectorTableHandle tableHandle) + { + SchemaTableName schemaTableName = ((HudiTableHandle) tableHandle).getSchemaTableName(); + Table table = metastore.getTable(schemaTableName.getSchemaName(), schemaTableName.getTableName()) + .orElseThrow(() -> new TableNotFoundException(schemaTableName)); + boolean managed = !isExternalTable(table); + Optional<String> location = table.getStorage().getOptionalLocation(); + + metastore.dropTable(schemaTableName.getSchemaName(), schemaTableName.getTableName(), managed); + + if (managed && location.isPresent()) { + // Done explicitly as well as through the metastore's deleteData flag, as the Delta Lake + // connector does: whether a metastore acts on that flag varies by implementation, and a + // managed table that keeps its data behind is a table whose name cannot be reused. + try { + fileSystemFactory.create(session).deleteDirectory(Location.of(location.get())); Review Comment: **minor:** For a MOR table hive-synced with `hoodie.datasource.hive_sync.create_managed_table=true`, `_ro` and `_rt` are both MANAGED and share the base path (`HMSDDLExecutor` sets both to `META_SYNC_BASE_PATH`), so `DROP TABLE t_ro` recursively deletes the data `t_rt` still reads. Niche, but before this PR DROP could not delete anything. Not blocking, but could we skip the explicit `deleteDirectory` when the table name differs from `hoodie.properties`' table name, or when the serde params carry `hoodie.query.as.ro.table=true`? ########## hudi-trino/src/main/java/io/trino/plugin/hudi/HudiTableInitializer.java: ########## @@ -0,0 +1,116 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hudi; + +import io.trino.filesystem.TrinoFileSystem; +import io.trino.plugin.hudi.storage.TrinoStorageConfiguration; +import io.trino.spi.TrinoException; +import io.trino.spi.connector.ConnectorTableMetadata; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.common.table.HoodieTableVersion; +import org.apache.hudi.storage.StoragePath; + +import java.io.IOException; +import java.util.List; +import java.util.Map; + +import static io.trino.plugin.hudi.HudiErrorCode.HUDI_META_CLIENT_ERROR; +import static io.trino.plugin.hudi.HudiTableProperties.getHiveStylePartitioning; +import static io.trino.plugin.hudi.HudiTableProperties.getHoodieProperties; +import static io.trino.plugin.hudi.HudiTableProperties.getKeyGeneratorClass; +import static io.trino.plugin.hudi.HudiTableProperties.getOrderingFields; +import static io.trino.plugin.hudi.HudiTableProperties.getPartitionedBy; +import static io.trino.plugin.hudi.HudiTableProperties.getPrimaryKey; +import static io.trino.plugin.hudi.HudiTableProperties.getRecordMergeMode; +import static io.trino.plugin.hudi.HudiTableProperties.getTableType; +import static java.lang.String.format; + +/** + * Writes a table's {@code .hoodie} directory. Pure hudi-common: no Spark, no write client and no + * metastore, so it can run before the catalog entry exists without entering any row-writing path. + */ +public final class HudiTableInitializer +{ + /** + * The {@code hoodie.table.version} written by this connector. + * <p> + * Pinned rather than inherited from {@link HoodieTableVersion#current()} so that a future bump of + * {@code current()} cannot silently change what the connector emits. Raising this should be a + * deliberate change with a test diff, not a side effect of upgrading Hudi; the assertion in + * {@code TestHudiDdl} fails when the two drift. + */ + public static final HoodieTableVersion CREATED_TABLE_VERSION = HoodieTableVersion.TEN; + + private HudiTableInitializer() {} + + /** + * Initializes storage for an empty table. + * + * @param tableSchema the schema that also produces the metastore column list, so the two cannot + * disagree + */ + public static void initializeTable( + TrinoFileSystem fileSystem, + String basePath, + ConnectorTableMetadata tableMetadata, + HoodieSchema tableSchema) + { + Map<String, Object> properties = tableMetadata.getProperties(); + HoodieTableMetaClient.TableBuilder builder = HoodieTableMetaClient.newTableBuilder() + .setTableType(getTableType(properties)) + .setTableName(tableMetadata.getTable().getTableName()) + .setDatabaseName(tableMetadata.getTable().getSchemaName()) + .setTableVersion(CREATED_TABLE_VERSION) + .setTableCreateSchema(tableSchema.toAvroSchema().toString()); + + List<String> primaryKey = getPrimaryKey(properties); + if (!primaryKey.isEmpty()) { + builder.setRecordKeyFields(String.join(",", primaryKey)); + } + List<String> partitionedBy = getPartitionedBy(properties); + if (!partitionedBy.isEmpty()) { + builder.setPartitionFields(String.join(",", partitionedBy)); Review Comment: **minor:** With `key_generator_class` set to `CustomKeyGenerator`/`CustomAvroKeyGenerator`, partition fields are written as bare names, but those key generators need `field:TYPE` (HUDI-7996, #11638). Spark SQL then falls back to the bare names in `ProvidesHoodieConfig.getPartitionPathFieldWriteConfig` ("This may fail the write operation"). Not blocking, but could `HudiTableValidation` reject the Custom key generators until there is a way to declare partition types? ########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiTableInitializer.java: ########## @@ -0,0 +1,200 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hudi; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import io.trino.filesystem.TrinoFileSystem; +import io.trino.filesystem.memory.MemoryFileSystem; +import io.trino.plugin.hudi.storage.HudiTrinoStorage; +import io.trino.plugin.hudi.storage.TrinoStorageConfiguration; +import io.trino.plugin.hudi.util.HudiSchemaConverter; +import io.trino.spi.connector.ColumnMetadata; +import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.SchemaTableName; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.table.HoodieTableConfig; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.storage.StoragePath; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Map; + +import static io.trino.plugin.hudi.HudiTableInitializer.CREATED_TABLE_VERSION; +import static io.trino.plugin.hudi.HudiTableProperties.ORDERING_FIELDS_PROPERTY; +import static io.trino.plugin.hudi.HudiTableProperties.PARTITIONED_BY_PROPERTY; +import static io.trino.plugin.hudi.HudiTableProperties.PRIMARY_KEY_PROPERTY; +import static io.trino.plugin.hudi.HudiTableProperties.TABLE_TYPE_PROPERTY; +import static io.trino.spi.type.BigintType.BIGINT; +import static io.trino.spi.type.TimestampType.createTimestampType; +import static io.trino.spi.type.VarcharType.VARCHAR; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Covers table initialization in isolation from the metastore, and in particular that it works at + * all through {@link HudiTrinoStorage}: the connector has to use the {@code HoodieStorage} overload + * of {@code initTable} because the configuration-based one resolves storage reflectively via a + * {@code (StoragePath, StorageConfiguration)} constructor that a session-scoped + * {@code TrinoFileSystem} cannot supply. + */ +final class TestHudiTableInitializer +{ + private static final String BASE_PATH = "memory:///warehouse/trips"; + + @Test + void testInitializesReadableTableMetadata() + { + TrinoFileSystem fileSystem = new MemoryFileSystem(); + initialize(fileSystem, HoodieTableType.COPY_ON_WRITE, ImmutableList.of("city")); + + HoodieTableConfig tableConfig = loadMetaClient(fileSystem).getTableConfig(); + assertThat(tableConfig.getTableType()).isEqualTo(HoodieTableType.COPY_ON_WRITE); + assertThat(tableConfig.getTableName()).isEqualTo("trips"); + assertThat(tableConfig.getRecordKeyFields().get()).containsExactly("id"); + assertThat(tableConfig.getPartitionFields().get()).containsExactly("city"); + } + + @Test + void testMergeOnReadIsInitializable() + { + TrinoFileSystem fileSystem = new MemoryFileSystem(); + initialize(fileSystem, HoodieTableType.MERGE_ON_READ, ImmutableList.of()); + + assertThat(loadMetaClient(fileSystem).getTableConfig().getTableType()) + .isEqualTo(HoodieTableType.MERGE_ON_READ); + } + + @Test + void testTableVersionIsPinnedNotInherited() + { + // Guards the pin: if HoodieTableVersion.current() moves ahead of CREATED_TABLE_VERSION, that + // is a deliberate decision and this assertion is where it has to be made. + TrinoFileSystem fileSystem = new MemoryFileSystem(); + initialize(fileSystem, HoodieTableType.COPY_ON_WRITE, ImmutableList.of()); + + assertThat(loadMetaClient(fileSystem).getTableConfig().getTableVersion()) + .isEqualTo(CREATED_TABLE_VERSION); Review Comment: **minor:** This compares against the production `CREATED_TABLE_VERSION`, so bumping the constant can never fail it, and the same holds for `TestHudiDdl` lines 133 and 164. The "Guards the pin" comment and the constant's javadoc ("the assertion in TestHudiDdl fails when the two drift") overstate it. Not blocking, but could at least one of these assert the literal `HoodieTableVersion.TEN`? ########## hudi-trino/src/main/java/io/trino/plugin/hudi/HudiMetadata.java: ########## @@ -304,6 +321,202 @@ public Optional<Object> getInfo(ConnectorSession session, ConnectorTableHandle t return Optional.of(new HudiTableInfo(table.getSchemaTableName(), table.getTableType().name(), table.getBasePath())); } + /** + * Creates an empty table: its {@code .hoodie} directory on storage and its catalog entry. + * <p> + * This is the single-call path only. {@code beginCreateTable}/{@code finishCreateTable} and a + * page sink are deliberately absent, so {@code CREATE TABLE AS} and {@code INSERT} still fail as + * unsupported; no rows are written here. + * <p> + * An explicit {@code location} makes the table external, matching Hudi's Spark SQL behaviour. An + * omitted one makes it managed, at {@code <schemaLocation>/<tableName>}, which is what decides + * whether {@code DROP TABLE} later deletes the data. + */ + @Override + public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, SaveMode saveMode) + { + SchemaTableName schemaTableName = tableMetadata.getTable(); + if (saveMode == SaveMode.REPLACE) { + // Replacing a table means deciding what happens to the rows already in it, which is the + // write path this connector does not yet have. + throw new TrinoException(NOT_SUPPORTED, "This connector does not support replacing tables"); + } + Database database = metastore.getDatabase(schemaTableName.getSchemaName()) + .orElseThrow(() -> new SchemaNotFoundException(schemaTableName.getSchemaName())); + if (metastore.getTable(schemaTableName.getSchemaName(), schemaTableName.getTableName()).isPresent()) { + if (saveMode == SaveMode.IGNORE) { + return; + } + throw new TrinoException(ALREADY_EXISTS, "Table already exists: " + schemaTableName); + } + + // Everything that can be rejected is rejected before storage is touched, so a bad statement + // leaves nothing behind. + HudiTableValidation.validateCreateTable(tableMetadata); + + Map<String, Object> properties = tableMetadata.getProperties(); + Optional<String> explicitLocation = getTableLocation(properties); + boolean external = explicitLocation.isPresent(); + String basePath = explicitLocation.orElseGet(() -> defaultTableLocation(database, schemaTableName)); + + TrinoFileSystem fileSystem = fileSystemFactory.create(session); + checkLocationIsEmpty(fileSystem, basePath); + + // One schema object produces both hoodie.table.create.schema and the metastore column list, + // so the two cannot disagree (HUDI-9435). + HoodieSchema tableSchema = HudiSchemaConverter.toTableSchema( + tableMetadata.getColumns(), schemaTableName.getTableName()); + Table table = HudiMetastoreTables.buildTable( + schemaTableName, + basePath, + getTableType(properties), + tableSchema, + getPartitionedBy(properties), + external, + Optional.of(session.getUser()), + tableMetadata.getComment()); + try { + HudiTableInitializer.initializeTable(fileSystem, basePath, tableMetadata, tableSchema); + metastore.createTable(table, NO_PRIVILEGES); + } + catch (TableAlreadyExistsException e) { + // Another CREATE TABLE may have initialized the same managed location and won the + // metastore race. Its catalog entry now owns .hoodie, so deleting it would corrupt the + // live table. A different location still belongs to this failed attempt and is safe to + // clean up. + if (!isTableRegisteredAtLocation(schemaTableName, basePath, e)) { + cleanupTableMetadata(fileSystem, basePath, e); + } + throw e; + } + catch (RuntimeException e) { + // Initialization may have written some or all of .hoodie, but no catalog entry from + // this call references it. Left there it would make a retry fail the emptiness check. + // + // Only .hoodie is removed, never the base path: the base path may have been created by + // someone else, and this connector did not create it. On object storage there is no + // directory to delete in any case -- deleteDirectory removes the objects under the + // prefix, which is exactly the set initTable wrote or may have partially written. + cleanupTableMetadata(fileSystem, basePath, e); + throw e; + } + } + + private boolean isTableRegisteredAtLocation(SchemaTableName tableName, String basePath, RuntimeException failure) + { + try { + return metastore.getTable(tableName.getSchemaName(), tableName.getTableName()) + .map(table -> table.getStorage().getLocation().equals(basePath)) + .orElse(false); + } + catch (RuntimeException lookupFailure) { + // When ownership cannot be established, leave storage intact. An orphan is recoverable; + // deleting metadata that a successful concurrent CREATE references is not. + failure.addSuppressed(lookupFailure); + return true; + } + } + + private static void cleanupTableMetadata(TrinoFileSystem fileSystem, String basePath, RuntimeException failure) + { + try { + fileSystem.deleteDirectory(Location.of(appendPath(basePath, METAFOLDER_NAME))); + } + catch (IOException | RuntimeException cleanupFailure) { + failure.addSuppressed(cleanupFailure); + } + } + + /** + * Drops the catalog entry, and the data too when the table is managed. + * <p> + * A table registered with an explicit location is external and its data outlives the catalog + * entry; {@code register_table} always produces such a table. Trino's {@code DROP TABLE} has no + * {@code PURGE} clause, so there is no way to ask for an external table's data to be deleted. + */ + @Override + public void dropTable(ConnectorSession session, ConnectorTableHandle tableHandle) + { + SchemaTableName schemaTableName = ((HudiTableHandle) tableHandle).getSchemaTableName(); + Table table = metastore.getTable(schemaTableName.getSchemaName(), schemaTableName.getTableName()) + .orElseThrow(() -> new TableNotFoundException(schemaTableName)); + boolean managed = !isExternalTable(table); + Optional<String> location = table.getStorage().getOptionalLocation(); + + metastore.dropTable(schemaTableName.getSchemaName(), schemaTableName.getTableName(), managed); + + if (managed && location.isPresent()) { + // Done explicitly as well as through the metastore's deleteData flag, as the Delta Lake + // connector does: whether a metastore acts on that flag varies by implementation, and a + // managed table that keeps its data behind is a table whose name cannot be reused. + try { + fileSystemFactory.create(session).deleteDirectory(Location.of(location.get())); + } + catch (IOException e) { + throw new TrinoException(HUDI_FILESYSTEM_ERROR, format( + "Failed to delete directory %s of the dropped table %s", location.get(), schemaTableName), e); + } + } + } + + /** + * Whether the metastore considers this table external, erring towards yes. + * <p> + * Hive records this twice, as the table type and as the {@code EXTERNAL} parameter, and they can + * disagree -- a table created by another tool may set only one. Treating either signal as + * decisive keeps {@code DROP TABLE} from deleting data it does not own; the opposite mistake is + * unrecoverable. + */ + private static boolean isExternalTable(Table table) + { + return EXTERNAL_TABLE.name().equals(table.getTableType()) + || "TRUE".equalsIgnoreCase(table.getParameters().getOrDefault("EXTERNAL", "")); + } + + /** + * Where a managed table's data goes: {@code <schemaLocation>/<tableName>}, as the Delta Lake + * connector derives it. + * <p> + * A schema with no location of its own has nowhere to put a managed table, and the error says so + * rather than reporting a null path further down. Schemas imported from external metastores can + * legitimately omit a location. + */ + private static String defaultTableLocation(Database database, SchemaTableName schemaTableName) + { + String schemaLocation = database.getLocation() + .filter(location -> !location.isEmpty()) + .orElseThrow(() -> new TrinoException(NOT_SUPPORTED, format( + "Schema '%s' has no location, so a managed table cannot be created in it: set the '%s' table property, or give the schema a location", + schemaTableName.getSchemaName(), LOCATION_PROPERTY))); + return appendPath(schemaLocation, escapeTableName(schemaTableName.getTableName())); + } + + /** + * Requires the target location to hold no files, for managed and external tables alike. + * <p> + * {@code initTable} writes {@code hoodie.properties}, which would overwrite the metadata of a + * Hudi table that already lives here; {@code register_table} is the way to adopt existing data. + * It also makes the rollback in {@link #createTable} exactly correct, since the location held + * nothing that this connector did not write. + * <p> + * This is a prefix listing, not a directory check, so it means the same thing on object storage + * -- where no directory exists to be inspected -- as it does on HDFS. + */ + private static void checkLocationIsEmpty(TrinoFileSystem fileSystem, String basePath) Review Comment: **minor:** None of these safety branches has a test: - this `already contains files` rejection (the guard against overwriting an existing `hoodie.properties`) - the generic cleanup catch in `createTable` - `isExternalTable` with only one external signal (Glue sync with `create_managed_table=true` gives EXTERNAL_TABLE without `EXTERNAL=TRUE`) - `unregister_table` on a plain Hive table - `record_merge_mode`, `key_generator_class`, `hive_style_partitioning`, and the `hoodie_properties` rejection path Not blocking, but could `TestHudiDdl` / `TestHudiMetadata` / `TestHudiSharedMetastore` get one case each, asserting data and metadata survive where relevant? ########## hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHoodieMetastoreTableDescriptorFormatNames.java: ########## @@ -0,0 +1,84 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hudi.hive; + +import org.apache.hudi.common.model.HoodieFileFormat; +import org.apache.hudi.hadoop.utils.HoodieInputFormatUtils; +import org.apache.hudi.sync.common.util.HoodieMetastoreTableDescriptor; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Pins {@link HoodieMetastoreTableDescriptor}'s format-name constants to the values hive-sync + * actually emits. + * + * <p>The constants exist because {@link HoodieInputFormatUtils} lives in {@code hudi-hadoop-mr} and + * pulls in Hadoop MapReduce, so a caller without that on its classpath cannot reach it. Repeating + * the names is safe only while something checks that the copies agree; this is that check, and it + * lives here because {@code hudi-hive-sync} is the nearest module that can see both. + * + * <p>If this fails, a format name changed on one side only. A Hudi table registered with an input + * format the reader does not recognise is a table that reader refuses to query at all, since the + * input format is the only record of the table type in the metastore. + */ +class TestHoodieMetastoreTableDescriptorFormatNames { + + @Test + void inputFormatNamesMatchHiveSync() { + assertEquals( + HoodieInputFormatUtils.getInputFormatClassName(HoodieFileFormat.PARQUET, false), + HoodieMetastoreTableDescriptor.PARQUET_INPUT_FORMAT_CLASS); + assertEquals( + HoodieInputFormatUtils.getInputFormatClassName(HoodieFileFormat.PARQUET, true), + HoodieMetastoreTableDescriptor.PARQUET_REALTIME_INPUT_FORMAT_CLASS); + } + + @Test + void outputFormatAndSerdeNamesMatchHiveSync() { + assertEquals( + HoodieInputFormatUtils.getOutputFormatClassName(HoodieFileFormat.PARQUET), + HoodieMetastoreTableDescriptor.PARQUET_OUTPUT_FORMAT_CLASS); + assertEquals( + HoodieInputFormatUtils.getSerDeClassName(HoodieFileFormat.PARQUET), + HoodieMetastoreTableDescriptor.PARQUET_SERDE_CLASS); + } + + @Test + void theDescriptorResolvesTheSameInputFormatHiveSyncWouldForEachTableType() { + // Copy-on-Write syncs one table with the non-realtime format; Merge-on-Read's snapshot view + // uses the realtime one. See HiveSyncTool#doSync. + assertEquals( + HoodieInputFormatUtils.getInputFormatClassName(HoodieFileFormat.PARQUET, false), + HoodieMetastoreTableDescriptor.inputFormatClassName(false)); + assertEquals( + HoodieInputFormatUtils.getInputFormatClassName(HoodieFileFormat.PARQUET, true), + HoodieMetastoreTableDescriptor.inputFormatClassName(true)); + } + + @Test + void theSchemaStringLengthThresholdMatchesTheHiveSyncDefault() { + // Keeping these equal is what makes a table registered through the descriptor byte-identical to Review Comment: **minor:** This class pins the class-name constants and the threshold default, but `HiveSyncTool` does not use the descriptor, so nothing checks that the parameters are actually byte-identical to what hive-sync writes. Not blocking, but could `TestHiveSyncTool`'s table-properties test (around line 757) also assert that `forSnapshotView(...).getTableParameters()` matches the synced `EXTERNAL` and `spark.sql.*` parameters? ########## hudi-trino/src/main/java/io/trino/plugin/hudi/procedure/RegisterTableProcedure.java: ########## @@ -0,0 +1,147 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hudi.procedure; + +import com.google.common.collect.ImmutableList; +import com.google.inject.Inject; +import com.google.inject.Provider; +import io.trino.filesystem.Location; +import io.trino.filesystem.TrinoFileSystem; +import io.trino.filesystem.TrinoFileSystemFactory; +import io.trino.metastore.HiveMetastore; +import io.trino.metastore.HiveMetastoreFactory; +import io.trino.plugin.hudi.HudiMetastoreTables; +import io.trino.plugin.hudi.HudiUtil; +import io.trino.spi.TrinoException; +import io.trino.spi.classloader.ThreadContextClassLoader; +import io.trino.spi.connector.ConnectorAccessControl; +import io.trino.spi.connector.ConnectorSession; +import io.trino.spi.connector.SchemaNotFoundException; +import io.trino.spi.connector.SchemaTableName; +import io.trino.spi.procedure.Procedure; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.table.HoodieTableMetaClient; + +import java.lang.invoke.MethodHandle; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import static io.trino.metastore.PrincipalPrivileges.NO_PRIVILEGES; +import static io.trino.plugin.base.util.Procedures.checkProcedureArgument; +import static io.trino.plugin.hudi.HudiTableProperties.LOCATION_PROPERTY; +import static io.trino.spi.StandardErrorCode.ALREADY_EXISTS; +import static io.trino.spi.type.VarcharType.VARCHAR; +import static java.lang.invoke.MethodHandles.lookup; +import static java.util.Objects.requireNonNull; + +public class RegisterTableProcedure + implements Provider<Procedure> +{ + private static final MethodHandle REGISTER_TABLE; + + static { + try { + REGISTER_TABLE = lookup().unreflect(RegisterTableProcedure.class.getMethod( + "registerTable", + ConnectorSession.class, + ConnectorAccessControl.class, + String.class, + String.class, + String.class)); + } + catch (ReflectiveOperationException e) { + throw new AssertionError(e); + } + } + + private final HiveMetastoreFactory metastoreFactory; + private final TrinoFileSystemFactory fileSystemFactory; + + @Inject + public RegisterTableProcedure(HiveMetastoreFactory metastoreFactory, TrinoFileSystemFactory fileSystemFactory) + { + this.metastoreFactory = requireNonNull(metastoreFactory, "metastoreFactory is null"); + this.fileSystemFactory = requireNonNull(fileSystemFactory, "fileSystemFactory is null"); + } + + @Override + public Procedure get() + { + return new Procedure( + "system", + "register_table", + ImmutableList.of( + new Procedure.Argument("SCHEMA_NAME", VARCHAR), + new Procedure.Argument("TABLE_NAME", VARCHAR), + new Procedure.Argument("TABLE_LOCATION", VARCHAR)), + REGISTER_TABLE.bindTo(this)); + } + + public void registerTable( + ConnectorSession session, + ConnectorAccessControl accessControl, + String schemaName, + String tableName, + String tableLocation) + { + try (ThreadContextClassLoader _ = new ThreadContextClassLoader(getClass().getClassLoader())) { + doRegisterTable(session, accessControl, schemaName, tableName, tableLocation); + } + } + + private void doRegisterTable( + ConnectorSession session, + ConnectorAccessControl accessControl, + String schemaName, + String tableName, + String tableLocation) + { + checkProcedureArgument(schemaName != null, "schema_name cannot be null"); + checkProcedureArgument(tableName != null, "table_name cannot be null"); + checkProcedureArgument(tableLocation != null, "table_location cannot be null"); + + SchemaTableName schemaTableName = new SchemaTableName(schemaName, tableName); + String basePath = Location.of(tableLocation).toString(); + HiveMetastore metastore = metastoreFactory.createMetastore(Optional.of(session.getIdentity())); + if (metastore.getDatabase(schemaName).isEmpty()) { Review Comment: **nit:** The metastore lookups use the raw `schemaName`/`tableName` while `createTable` gets the lowercased `schemaTableName`, and the existence checks run before `checkCanCreateTable`, which tells a user without access whether a table exists (Delta's `RegisterTableProcedure` checks access first). Same in `UnregisterTableProcedure`. Feel free to ignore, but could we use `schemaTableName` throughout and move the access check up? ########## hudi-trino/src/test/java/io/trino/plugin/hudi/TestHudiTableInitializer.java: ########## @@ -0,0 +1,200 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hudi; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import io.trino.filesystem.TrinoFileSystem; +import io.trino.filesystem.memory.MemoryFileSystem; +import io.trino.plugin.hudi.storage.HudiTrinoStorage; +import io.trino.plugin.hudi.storage.TrinoStorageConfiguration; +import io.trino.plugin.hudi.util.HudiSchemaConverter; +import io.trino.spi.connector.ColumnMetadata; +import io.trino.spi.connector.ConnectorTableMetadata; +import io.trino.spi.connector.SchemaTableName; +import org.apache.hudi.common.model.HoodieTableType; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.table.HoodieTableConfig; +import org.apache.hudi.common.table.HoodieTableMetaClient; +import org.apache.hudi.storage.StoragePath; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Map; + +import static io.trino.plugin.hudi.HudiTableInitializer.CREATED_TABLE_VERSION; +import static io.trino.plugin.hudi.HudiTableProperties.ORDERING_FIELDS_PROPERTY; +import static io.trino.plugin.hudi.HudiTableProperties.PARTITIONED_BY_PROPERTY; +import static io.trino.plugin.hudi.HudiTableProperties.PRIMARY_KEY_PROPERTY; +import static io.trino.plugin.hudi.HudiTableProperties.TABLE_TYPE_PROPERTY; +import static io.trino.spi.type.BigintType.BIGINT; +import static io.trino.spi.type.TimestampType.createTimestampType; +import static io.trino.spi.type.VarcharType.VARCHAR; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Covers table initialization in isolation from the metastore, and in particular that it works at + * all through {@link HudiTrinoStorage}: the connector has to use the {@code HoodieStorage} overload + * of {@code initTable} because the configuration-based one resolves storage reflectively via a + * {@code (StoragePath, StorageConfiguration)} constructor that a session-scoped + * {@code TrinoFileSystem} cannot supply. + */ +final class TestHudiTableInitializer +{ + private static final String BASE_PATH = "memory:///warehouse/trips"; + + @Test + void testInitializesReadableTableMetadata() + { + TrinoFileSystem fileSystem = new MemoryFileSystem(); + initialize(fileSystem, HoodieTableType.COPY_ON_WRITE, ImmutableList.of("city")); + + HoodieTableConfig tableConfig = loadMetaClient(fileSystem).getTableConfig(); + assertThat(tableConfig.getTableType()).isEqualTo(HoodieTableType.COPY_ON_WRITE); + assertThat(tableConfig.getTableName()).isEqualTo("trips"); + assertThat(tableConfig.getRecordKeyFields().get()).containsExactly("id"); + assertThat(tableConfig.getPartitionFields().get()).containsExactly("city"); + } + + @Test + void testMergeOnReadIsInitializable() + { + TrinoFileSystem fileSystem = new MemoryFileSystem(); + initialize(fileSystem, HoodieTableType.MERGE_ON_READ, ImmutableList.of()); + + assertThat(loadMetaClient(fileSystem).getTableConfig().getTableType()) + .isEqualTo(HoodieTableType.MERGE_ON_READ); + } + + @Test + void testTableVersionIsPinnedNotInherited() + { + // Guards the pin: if HoodieTableVersion.current() moves ahead of CREATED_TABLE_VERSION, that + // is a deliberate decision and this assertion is where it has to be made. + TrinoFileSystem fileSystem = new MemoryFileSystem(); + initialize(fileSystem, HoodieTableType.COPY_ON_WRITE, ImmutableList.of()); + + assertThat(loadMetaClient(fileSystem).getTableConfig().getTableVersion()) + .isEqualTo(CREATED_TABLE_VERSION); + } + + @Test + void testCreateSchemaCarriesMetaFieldsAndDataColumns() + { + TrinoFileSystem fileSystem = new MemoryFileSystem(); + initialize(fileSystem, HoodieTableType.COPY_ON_WRITE, ImmutableList.of("city")); + + String createSchema = loadMetaClient(fileSystem).getTableConfig() + .getString(HoodieTableConfig.CREATE_SCHEMA); + assertThat(createSchema) + .contains("_hoodie_commit_time") + .contains("_hoodie_record_key") + .contains("\"name\":\"id\"") + .contains("\"name\":\"city\""); + } + + @Test + void testOrderingFieldsAreWrittenUnderTheCurrentKey() + { + // hoodie.table.precombine.field is only a deprecated alternative of + // hoodie.table.ordering.fields; a table created now must use the current key. + TrinoFileSystem fileSystem = new MemoryFileSystem(); + initialize(fileSystem, HoodieTableType.MERGE_ON_READ, ImmutableList.of()); + + HoodieTableConfig tableConfig = loadMetaClient(fileSystem).getTableConfig(); + assertThat(tableConfig.getString(HoodieTableConfig.ORDERING_FIELDS)).isEqualTo("event_time"); + } + + @Test + void testHoodiePassthroughReachesTableConfig() + { + // Only configs in HoodieTableConfig.PERSISTED_CONFIG_LIST survive TableBuilder#set; the + // property validator rejects anything else rather than letting it vanish here. + TrinoFileSystem fileSystem = new MemoryFileSystem(); + HudiTableInitializer.initializeTable( + fileSystem, + BASE_PATH, + tableMetadata(HoodieTableType.COPY_ON_WRITE, ImmutableList.of(), ImmutableMap.of( + "hoodie.keygen.timebased.timestamp.type", "DATE_STRING", + "hoodie.keygen.timebased.output.dateformat", "yyyy/MM/dd")), + schema()); + + HoodieTableConfig tableConfig = loadMetaClient(fileSystem).getTableConfig(); + assertThat(tableConfig.getString("hoodie.keygen.timebased.timestamp.type")).isEqualTo("DATE_STRING"); + assertThat(tableConfig.getString("hoodie.keygen.timebased.output.dateformat")).isEqualTo("yyyy/MM/dd"); + } + + @Test + void testInitGoesThroughHudisPluggableStorageExtensionPoint() Review Comment: **minor:** This test never calls `HudiTableInitializer`; it repeats `TestHudiTrinoStorage#testConfigurationWithoutFileSystemFailsClearly` (line 171) and `#testInitTableWritesThroughExtensionPoint` (line 206). The class javadoc also says the connector uses the `HoodieStorage` overload of `initTable`, but `HudiTableInitializer` calls the `StorageConfiguration` one. Not blocking, but could we drop this method and correct the javadoc? ########## hudi-trino/src/main/java/io/trino/plugin/hudi/util/HudiSchemaConverter.java: ########## @@ -0,0 +1,283 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.trino.plugin.hudi.util; + +import com.google.common.collect.ImmutableList; +import io.trino.spi.TrinoException; +import io.trino.spi.connector.ColumnMetadata; +import io.trino.spi.type.ArrayType; +import io.trino.spi.type.CharType; +import io.trino.spi.type.DecimalType; +import io.trino.spi.type.MapType; +import io.trino.spi.type.RowType; +import io.trino.spi.type.TimeType; +import io.trino.spi.type.TimestampType; +import io.trino.spi.type.TimestampWithTimeZoneType; +import io.trino.spi.type.Type; +import io.trino.spi.type.VarbinaryType; +import io.trino.spi.type.VarcharType; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.schema.HoodieSchemaField; +import org.apache.hudi.common.schema.HoodieSchemaType; +import org.apache.hudi.common.schema.HoodieSchemaUtils; + +import java.util.HashSet; +import java.util.List; +import java.util.Locale; +import java.util.Set; + +import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED; +import static io.trino.spi.type.BigintType.BIGINT; +import static io.trino.spi.type.BooleanType.BOOLEAN; +import static io.trino.spi.type.DateType.DATE; +import static io.trino.spi.type.DoubleType.DOUBLE; +import static io.trino.spi.type.IntegerType.INTEGER; +import static io.trino.spi.type.RealType.REAL; +import static io.trino.spi.type.SmallintType.SMALLINT; +import static io.trino.spi.type.TinyintType.TINYINT; +import static io.trino.spi.type.UuidType.UUID; +import static java.lang.String.format; + +/** + * Converts a Trino column list into the {@link HoodieSchema} that becomes a table's + * {@code hoodie.table.create.schema}. + * <p> + * This direction did not previously exist in the connector: the read path only maps Hudi/Avro + * types into Trino types. The mapping here is the inverse of Trino's own + * {@code NativeLogicalTypesAvroTypeManager}, so a column created through this converter reads back + * as the same Trino type -- except where a mapping is deliberately widening, noted per case below. + * <p> + * Mappings that widen, so a column does not read back as the type it was declared with: + * <ul> + * <li>{@code TINYINT} and {@code SMALLINT} become Avro {@code int} and read back as + * {@code INTEGER}. Avro has no narrower integer, and Spark's Avro conversion widens the same + * way, so rejecting them would make the connector stricter than its peers for no gain.</li> + * <li>{@code VARCHAR(n)} becomes Avro {@code string} and reads back as unbounded {@code VARCHAR}. + * Avro strings carry no length bound; nothing enforces {@code n} once another engine writes.</li> + * <li>{@code TIMESTAMP(p)} for a {@code p} that is not exactly 3 or 6 rounds up to the next + * representable precision, since Avro offers only millisecond and microsecond logical types. + * The widening is lossless. Separately, the Hive Metastore's {@code timestamp} carries no + * precision at all, so every timestamp column reads back at the precision the connector + * requests from the metastore rather than the one it was declared with.</li> + * </ul> + * Types rejected outright, because the alternative is a mapping that is quietly wrong rather than + * merely wider: + * <ul> + * <li>{@code CHAR(n)} -- Avro has no fixed-width string, so the blank-padding semantics that + * distinguish {@code CHAR} from {@code VARCHAR} would be silently dropped.</li> + * <li>{@code TIMESTAMP(p) WITH TIME ZONE} -- Avro's timestamp logical types carry an instant, not + * an instant plus a zone, so the per-value zone would be lost.</li> + * <li>{@code TIMESTAMP(p)} beyond microsecond precision -- Avro's nanosecond logical types are not + * among those the connector's read path decodes, so such a column would be written and then be + * unreadable.</li> + * <li>{@code UUID} and {@code TIME(p)} -- both have a faithful Avro logical type, but neither has a + * Hive counterpart, so {@code HiveTypeTranslator#toHiveType} rejects them and the column could + * never be registered in the metastore. Since {@code HudiMetadata#getColumnHandles} reads + * columns from the metastore, such a column would also be unreadable. Rejecting here keeps the + * failure at the column that caused it.</li> + * <li>{@code MAP} with a non-{@code VARCHAR} key type -- Avro map keys are always strings.</li> + * <li>Unnamed {@code ROW} fields -- Avro record fields must be named.</li> + * </ul> + * {@code DECIMAL} maps to Avro {@code bytes} rather than {@code fixed}. Both are decodable by the + * read path, and {@code bytes} is what Hudi's own {@link HoodieSchema#createDecimal(int, int)} + * helper produces; {@code fixed} would additionally require inventing a unique schema name per + * decimal column, since Avro fixed types are named and must not collide within one schema. + * <p> + * Nullability: Trino carries nullability per column but not per element inside a {@code ROW}, + * {@code ARRAY} or {@code MAP}. A column's own {@link ColumnMetadata#isNullable()} is honoured at + * the top level; everything nested is made nullable, which is what Spark's Avro conversion also + * produces. + */ +public final class HudiSchemaConverter +{ + private static final String NAMESPACE = "hoodie.trino"; + private static final int MAX_MILLIS_PRECISION = 3; + private static final int MAX_MICROS_PRECISION = 6; + + private HudiSchemaConverter() {} + + /** + * Builds the table schema, with Hudi's five meta fields prepended exactly as + * {@link HoodieSchemaUtils#addMetadataFields} would for any other engine. + * <p> + * The returned schema is the single source of truth for both {@code hoodie.table.create.schema} + * and the Hive Metastore column list. Deriving those two from separate inputs is what produces a + * table whose metastore descriptor and Hudi schema disagree (HUDI-9435). + * + * @param columns all table columns in declaration order, partition columns included + * @param tableName used to name the Avro record + */ + public static HoodieSchema toTableSchema(List<ColumnMetadata> columns, String tableName) + { + String recordName = sanitizeName(tableName); + RecordNameAllocator recordNames = new RecordNameAllocator(recordName); + ImmutableList.Builder<HoodieSchemaField> fields = ImmutableList.builder(); + for (ColumnMetadata column : columns) { + HoodieSchema fieldSchema = toHoodieSchema(column.getType(), column.getName(), recordNames); + if (column.isNullable()) { + fields.add(HoodieSchemaField.of( + column.getName(), + HoodieSchema.createNullable(fieldSchema), + column.getComment().orElse(null), + HoodieSchema.NULL_VALUE)); + } + else { + fields.add(HoodieSchemaField.of(column.getName(), fieldSchema, column.getComment().orElse(null), null)); + } + } + HoodieSchema record = HoodieSchema.createRecord( + recordName, NAMESPACE, null, fields.build()); + return HoodieSchemaUtils.addMetadataFields(record); + } + + /** + * Maps a single Trino type, failing with a message that names the type when no faithful Avro + * representation exists. {@code path} identifies the column (and nested field, for a + * {@code ROW}) so an error points at the offending column and so nested records get distinct + * Avro names. + */ + public static HoodieSchema toHoodieSchema(Type type, String path) + { + return toHoodieSchema(type, path, new RecordNameAllocator()); + } + + private static HoodieSchema toHoodieSchema(Type type, String path, RecordNameAllocator recordNames) + { + if (BOOLEAN.equals(type)) { + return HoodieSchema.create(HoodieSchemaType.BOOLEAN); + } + // Avro's narrowest integer is int; both widen, and read back as INTEGER. + if (TINYINT.equals(type) || SMALLINT.equals(type) || INTEGER.equals(type)) { + return HoodieSchema.create(HoodieSchemaType.INT); + } + if (BIGINT.equals(type)) { + return HoodieSchema.create(HoodieSchemaType.LONG); + } + if (REAL.equals(type)) { + return HoodieSchema.create(HoodieSchemaType.FLOAT); + } + if (DOUBLE.equals(type)) { + return HoodieSchema.create(HoodieSchemaType.DOUBLE); + } + if (DATE.equals(type)) { + return HoodieSchema.createDate(); + } + if (UUID.equals(type)) { + throw unsupported(type, path, "the Hive Metastore has no UUID type, so the column could not be registered in the catalog; use VARCHAR"); + } + if (type instanceof VarbinaryType) { + return HoodieSchema.create(HoodieSchemaType.BYTES); + } + if (type instanceof DecimalType decimalType) { + return HoodieSchema.createDecimal(decimalType.getPrecision(), decimalType.getScale()); + } + if (type instanceof VarcharType) { + // Any length bound is dropped; Avro strings are unbounded. + return HoodieSchema.create(HoodieSchemaType.STRING); + } + if (type instanceof CharType) { + throw unsupported(type, path, "Avro has no fixed-width string type, so CHAR padding semantics would be lost; use VARCHAR"); + } + if (type instanceof TimeType) { + throw unsupported(type, path, "the Hive Metastore has no TIME type, so the column could not be registered in the catalog"); + } + if (type instanceof TimestampType timestampType) { + int precision = timestampType.getPrecision(); + if (precision <= MAX_MILLIS_PRECISION) { + return HoodieSchema.createTimestampMillis(); + } + if (precision <= MAX_MICROS_PRECISION) { + return HoodieSchema.createTimestampMicros(); + } + throw unsupported(type, path, format( + "Avro timestamp logical types stop at microsecond precision; TIMESTAMP(%s) or narrower is supported", + MAX_MICROS_PRECISION)); + } + if (type instanceof TimestampWithTimeZoneType) { + throw unsupported(type, path, "Avro timestamp logical types carry an instant but no zone, so the per-value time zone would be lost; use TIMESTAMP without time zone"); + } + if (type instanceof ArrayType arrayType) { + return HoodieSchema.createArray( + HoodieSchema.createNullable(toHoodieSchema(arrayType.getElementType(), path + "_element", recordNames))); + } + if (type instanceof MapType mapType) { + if (!(mapType.getKeyType() instanceof VarcharType)) { + throw unsupported(type, path, format( + "Avro map keys are always strings, but the key type is %s; use a VARCHAR key", + mapType.getKeyType().getDisplayName())); + } + return HoodieSchema.createMap( + HoodieSchema.createNullable(toHoodieSchema(mapType.getValueType(), path + "_value", recordNames))); + } + if (type instanceof RowType rowType) { + ImmutableList.Builder<HoodieSchemaField> fields = ImmutableList.builder(); + for (RowType.Field field : rowType.getFields()) { + String fieldName = field.getName() + .orElseThrow(() -> unsupported(type, path, "Avro record fields must be named, but this ROW has an unnamed field")); + fields.add(HoodieSchemaField.of( + fieldName, + HoodieSchema.createNullable(toHoodieSchema(field.getType(), path + "_" + fieldName, recordNames)), + null, + HoodieSchema.NULL_VALUE)); + } + return HoodieSchema.createRecord(recordNames.allocate(path), NAMESPACE, null, fields.build()); + } + throw unsupported(type, path, "the Hudi connector has no Avro mapping for this type"); + } + + private static TrinoException unsupported(Type type, String path, String reason) + { + return new TrinoException(NOT_SUPPORTED, format( + "Cannot create a Hudi table with column '%s' of type %s: %s", + path, type.getDisplayName(), reason)); + } + + /** + * Avro names must match {@code [A-Za-z_][A-Za-z0-9_]*}. Trino identifiers are already + * lower-cased and permit characters Avro does not, so anything else becomes an underscore. + */ + private static String sanitizeName(String name) Review Comment: **nit:** This re-implements `HoodieSchemaUtils.sanitizeName`, and the top-level record uses namespace `hoodie.trino` rather than `HoodieSchemaUtils.getRecordQualifiedName(tableName)`, the convention Flink aligned to in HUDI-6145 (#8587). Feel free to ignore, but could this reuse both helpers? ########## hudi-trino/src/main/java/io/trino/plugin/hudi/HudiMetadata.java: ########## @@ -304,6 +321,202 @@ public Optional<Object> getInfo(ConnectorSession session, ConnectorTableHandle t return Optional.of(new HudiTableInfo(table.getSchemaTableName(), table.getTableType().name(), table.getBasePath())); } + /** + * Creates an empty table: its {@code .hoodie} directory on storage and its catalog entry. + * <p> + * This is the single-call path only. {@code beginCreateTable}/{@code finishCreateTable} and a + * page sink are deliberately absent, so {@code CREATE TABLE AS} and {@code INSERT} still fail as + * unsupported; no rows are written here. + * <p> + * An explicit {@code location} makes the table external, matching Hudi's Spark SQL behaviour. An + * omitted one makes it managed, at {@code <schemaLocation>/<tableName>}, which is what decides + * whether {@code DROP TABLE} later deletes the data. + */ + @Override + public void createTable(ConnectorSession session, ConnectorTableMetadata tableMetadata, SaveMode saveMode) + { + SchemaTableName schemaTableName = tableMetadata.getTable(); + if (saveMode == SaveMode.REPLACE) { + // Replacing a table means deciding what happens to the rows already in it, which is the + // write path this connector does not yet have. + throw new TrinoException(NOT_SUPPORTED, "This connector does not support replacing tables"); + } + Database database = metastore.getDatabase(schemaTableName.getSchemaName()) + .orElseThrow(() -> new SchemaNotFoundException(schemaTableName.getSchemaName())); + if (metastore.getTable(schemaTableName.getSchemaName(), schemaTableName.getTableName()).isPresent()) { + if (saveMode == SaveMode.IGNORE) { + return; + } + throw new TrinoException(ALREADY_EXISTS, "Table already exists: " + schemaTableName); + } + + // Everything that can be rejected is rejected before storage is touched, so a bad statement + // leaves nothing behind. + HudiTableValidation.validateCreateTable(tableMetadata); + + Map<String, Object> properties = tableMetadata.getProperties(); + Optional<String> explicitLocation = getTableLocation(properties); + boolean external = explicitLocation.isPresent(); + String basePath = explicitLocation.orElseGet(() -> defaultTableLocation(database, schemaTableName)); + + TrinoFileSystem fileSystem = fileSystemFactory.create(session); + checkLocationIsEmpty(fileSystem, basePath); + + // One schema object produces both hoodie.table.create.schema and the metastore column list, + // so the two cannot disagree (HUDI-9435). + HoodieSchema tableSchema = HudiSchemaConverter.toTableSchema( + tableMetadata.getColumns(), schemaTableName.getTableName()); + Table table = HudiMetastoreTables.buildTable( + schemaTableName, + basePath, + getTableType(properties), + tableSchema, + getPartitionedBy(properties), + external, + Optional.of(session.getUser()), + tableMetadata.getComment()); + try { + HudiTableInitializer.initializeTable(fileSystem, basePath, tableMetadata, tableSchema); + metastore.createTable(table, NO_PRIVILEGES); + } + catch (TableAlreadyExistsException e) { + // Another CREATE TABLE may have initialized the same managed location and won the + // metastore race. Its catalog entry now owns .hoodie, so deleting it would corrupt the + // live table. A different location still belongs to this failed attempt and is safe to + // clean up. + if (!isTableRegisteredAtLocation(schemaTableName, basePath, e)) { + cleanupTableMetadata(fileSystem, basePath, e); + } + throw e; + } + catch (RuntimeException e) { + // Initialization may have written some or all of .hoodie, but no catalog entry from + // this call references it. Left there it would make a retry fail the emptiness check. + // + // Only .hoodie is removed, never the base path: the base path may have been created by + // someone else, and this connector did not create it. On object storage there is no + // directory to delete in any case -- deleteDirectory removes the objects under the + // prefix, which is exactly the set initTable wrote or may have partially written. + cleanupTableMetadata(fileSystem, basePath, e); + throw e; + } + } + + private boolean isTableRegisteredAtLocation(SchemaTableName tableName, String basePath, RuntimeException failure) + { + try { + return metastore.getTable(tableName.getSchemaName(), tableName.getTableName()) + .map(table -> table.getStorage().getLocation().equals(basePath)) Review Comment: **major:** This can also trigger without a concurrent peer. `ThriftHiveMetastore.createTable` retries transient errors and maps `AlreadyExistsException` to `TableAlreadyExistsException`, so a timeout after HMS already committed lands in this branch for this query's own table. With `location='s3://b/t/'` (trailing slash), HMS stores the normalized path, the equality fails, and the query deletes its own `.hoodie`. Could we compare normalized `Location`s, or stamp `trino_query_id` into the table parameters and decide ownership on that, as `DeltaLakeMetadata` does? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
