hudi-agent commented on code in PR #20012: URL: https://github.com/apache/hudi/pull/20012#discussion_r4068849764
########## hudi-trino/src/main/java/io/trino/plugin/hudi/util/HudiSchemaConverter.java: ########## @@ -0,0 +1,254 @@ +/* + * 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.List; +import java.util.Locale; + +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) + { + ImmutableList.Builder<HoodieSchemaField> fields = ImmutableList.builder(); + for (ColumnMetadata column : columns) { + HoodieSchema fieldSchema = toHoodieSchema(column.getType(), column.getName()); + 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( + sanitizeName(tableName), 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) + { + 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"))); + } + 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"))); + } + 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)), + null, + HoodieSchema.NULL_VALUE)); + } + return HoodieSchema.createRecord(sanitizeName(path), NAMESPACE, null, fields.build()); Review Comment: 🤖 Nested ROW records and the top-level record share the `hoodie.trino` namespace and are named purely from the sanitized path, so a table `trips` with a ROW column `trips`, or columns `a` (ROW with a ROW field `b`) and `a_b` (ROW), produce two distinct records with the same Avro fullname. `tableSchema.toAvroSchema().toString()` in `HudiTableInitializer` then throws `SchemaParseException: Can't redefine: hoodie.trino.a_b`. Could nested records get a distinct namespace (e.g. `hoodie.trino.<table>`) or a uniqueness check with a clear error? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/util/HoodieMetastoreTableDescriptor.java: ########## @@ -0,0 +1,324 @@ +/* + * 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, external, + useRealtimeInputFormat, false, "", DEFAULT_SCHEMA_STRING_LENGTH_THRESHOLD, false, + Collections.emptyMap()); + } + + /** + * The general form, retaining every knob hive-sync varies. Callers registering one table should + * prefer {@link #forSnapshotView}. + * + * @param useRealtimeInputFormat whether to use the real-time input format. Independent of + * {@code tableType} only because hive-sync registers a Merge-on-Read table twice, once each + * way; for Copy-on-Write it must be false, as there is no real-time view to read. + * @param readAsOptimized recorded as {@code hoodie.query.as.ro.table}. True only for the + * read-optimized view of a Merge-on-Read table. + * @param sparkVersion written as {@code spark.sql.create.version}; empty omits the property, as + * hive-sync does by default + * @param includeFieldDocs whether column comments are carried into the serialized Spark schema + * @param extraTableParameters additional table parameters, applied last so a caller can override + * anything derived here + */ + public static HoodieMetastoreTableDescriptor forView( Review Comment: 🤖 nit: forView takes four booleans plus a String and an int in a row (external, useRealtimeInputFormat, readAsOptimized, ..., includeFieldDocs) - easy for a caller to transpose two of them without the compiler catching it. Worth considering a small builder or parameter object for the general form instead of positional booleans. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-trino/src/main/java/io/trino/plugin/hudi/HudiMetadata.java: ########## @@ -304,6 +320,172 @@ 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 (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. + try { + fileSystem.deleteDirectory(Location.of(appendPath(basePath, METAFOLDER_NAME))); Review Comment: 🤖 If two sessions run `CREATE TABLE t (...)` concurrently for the same managed name, both can pass `checkLocationIsEmpty` before either `initTable` runs, both write `.hoodie` at the same location, and the one that loses `metastore.createTable` with ALREADY_EXISTS will delete the `.hoodie` the winner is now registered against. Would it make sense to skip the storage cleanup when the failure is a `TableAlreadyExistsException` (or re-check the location is still what this call wrote), so a concurrent DDL race can't strip a live table's metadata? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-sync/hudi-sync-common/src/main/java/org/apache/hudi/sync/common/util/HoodieMetastoreTableDescriptor.java: ########## @@ -0,0 +1,324 @@ +/* + * 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, external, + useRealtimeInputFormat, false, "", DEFAULT_SCHEMA_STRING_LENGTH_THRESHOLD, false, + Collections.emptyMap()); + } + + /** + * The general form, retaining every knob hive-sync varies. Callers registering one table should + * prefer {@link #forSnapshotView}. + * + * @param useRealtimeInputFormat whether to use the real-time input format. Independent of + * {@code tableType} only because hive-sync registers a Merge-on-Read table twice, once each + * way; for Copy-on-Write it must be false, as there is no real-time view to read. + * @param readAsOptimized recorded as {@code hoodie.query.as.ro.table}. True only for the + * read-optimized view of a Merge-on-Read table. + * @param sparkVersion written as {@code spark.sql.create.version}; empty omits the property, as + * hive-sync does by default + * @param includeFieldDocs whether column comments are carried into the serialized Spark schema + * @param extraTableParameters additional table parameters, applied last so a caller can override + * anything derived here + */ + public static HoodieMetastoreTableDescriptor forView( + HoodieSchema tableSchema, + List<String> partitionFieldNames, + HoodieTableType tableType, + String basePath, + boolean external, + boolean useRealtimeInputFormat, + boolean readAsOptimized, + String sparkVersion, + int schemaStringLengthThreshold, + boolean includeFieldDocs, + Map<String, String> extraTableParameters) { + if (tableSchema == null) { + throw new IllegalArgumentException("tableSchema is required"); + } + if (basePath == null || basePath.isEmpty()) { + throw new IllegalArgumentException("basePath is required"); + } + if (tableType == null) { + throw new IllegalArgumentException("tableType is required"); + } + if (useRealtimeInputFormat && tableType == HoodieTableType.COPY_ON_WRITE) { + throw new IllegalArgumentException( + "A Copy-on-Write table has no real-time view, so it cannot use the real-time input format"); + } + List<String> partitionNames = partitionFieldNames == null + ? Collections.emptyList() : new ArrayList<>(partitionFieldNames); + + Map<String, HoodieSchemaField> fieldsByName = new HashMap<>(); + for (HoodieSchemaField field : tableSchema.getFields()) { + fieldsByName.put(field.name(), field); + } + + // Partition columns first, in the order given, so the metastore's partition key order matches the + // partition path. A metastore reorders neither list, so this order is the table's for good. + List<HoodieSchemaField> partitionColumns = new ArrayList<>(partitionNames.size()); + Set<String> seenPartitionNames = new HashSet<>(); + for (String partitionName : partitionNames) { + if (!seenPartitionNames.add(partitionName)) { + throw new IllegalArgumentException("Partition column '" + partitionName + "' is listed more than once"); + } + HoodieSchemaField field = fieldsByName.get(partitionName); + if (field == null) { + // Not an error: see the parameter documentation on forSnapshotView. + field = HoodieSchemaField.of(partitionName, HoodieSchema.create(HoodieSchemaType.STRING)); + } + partitionColumns.add(field); + } + + // Everything else, in schema order, so the meta fields keep the leading positions + // addMetadataFields gave them. + List<HoodieSchemaField> dataColumns = new ArrayList<>(); + for (HoodieSchemaField field : tableSchema.getFields()) { + if (!seenPartitionNames.contains(field.name())) { + dataColumns.add(field); + } + } + + Map<String, String> tableParameters = new LinkedHashMap<>(); + if (external) { + tableParameters.put(EXTERNAL_PARAMETER, EXTERNAL_PARAMETER_VALUE); + } + // Emitted unconditionally. hive-sync gates these on + // hoodie.datasource.hive_sync.sync_as_datasource, which defaults to true; a table without + // spark.sql.sources.provider is not recognised by Spark SQL as a Hudi datasource table at all, + // so there is no reason for a newly created table to omit them. + tableParameters.putAll(SparkDataSourceTableUtils.getSparkTableProperties( + partitionNames, sparkVersion, schemaStringLengthThreshold, tableSchema, includeFieldDocs)); + if (extraTableParameters != null) { + tableParameters.putAll(extraTableParameters); + } + + Map<String, String> serdeParameters = new LinkedHashMap<>( + SparkDataSourceTableUtils.getSparkSerdeProperties(readAsOptimized, basePath)); + serdeParameters.put(SERIALIZATION_FORMAT_PARAMETER, SERIALIZATION_FORMAT_VALUE); + + return new HoodieMetastoreTableDescriptor( + dataColumns, + partitionColumns, + inputFormatClassName(tableType, useRealtimeInputFormat), + PARQUET_OUTPUT_FORMAT_CLASS, + PARQUET_SERDE_CLASS, + tableParameters, + serdeParameters, + external); + } + + /** + * The input format that records this table type. Equivalent to + * {@code HoodieInputFormatUtils#getInputFormatClassName(HoodieFileFormat.PARQUET, realtime)}, + * which lives in {@code hudi-hadoop-mr} and drags in Hadoop MapReduce. + */ + public static String inputFormatClassName(HoodieTableType tableType, boolean useRealtimeInputFormat) { Review Comment: 🤖 nit: tableType isn't used in the method body here (only useRealtimeInputFormat decides the result) - could this drop the unused parameter? A caller could reasonably assume this validates the tableType/realtime combination, but that check actually lives in forView. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> -- 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]
