924060929 commented on code in PR #64446:
URL: https://github.com/apache/doris/pull/64446#discussion_r3548321024
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogFactory.java:
##########
@@ -46,10 +45,10 @@ public class CatalogFactory {
private static final Logger LOG =
LogManager.getLogger(CatalogFactory.class);
// Only these catalog types are routed through the SPI connector path.
- // Other types (hms, iceberg, paimon, hudi) still use
+ // Other types (hms, iceberg, hudi) still use
// their built-in ExternalCatalog implementations until their
ConnectorProviders are fully ready.
private static final Set<String> SPI_READY_TYPES =
- ImmutableSet.of("jdbc", "es", "trino-connector", "max_compute");
+ ImmutableSet.of("jdbc", "es", "trino-connector", "max_compute",
"paimon");
Review Comment:
Adding `paimon` here is the line that flips paimon EXPLAIN output from
`VPAIMON_SCAN_NODE` to `VPluginDrivenScanNode` (verified live on this branch;
master PaimonScanNode.java:159 passes "PAIMON_SCAN_NODE"). The tracking issue's
invariant #3 explicitly lists EXPLAIN output as preserved, and
plan-doc/deviations-log has no entry for the header rename (the fix-e work
restored the explain *body* lines byte-for-byte but accepted the renamed
header). If the unified name is intentional, suggest recording it as a
deviation on #65185; if not, `planNodeName = connectorType.toUpperCase() +
"_SCAN_NODE"` in PluginDrivenScanNode's ctor preserves the legacy header for
every migrated connector at one stroke.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java:
##########
@@ -142,57 +396,183 @@ public List<ConnectorScanRange> planScan(
List<ConnectorScanRange> ranges = new ArrayList<>();
+ // Read the cpp-reader flag once: it selects the JNI split
serialization format (see encodeSplit).
+ boolean cppReader = isCppReaderEnabled(session);
+
+ // FIX-REST-VENDED-URI-NORMALIZE (P9-1): extract the per-table vended
token ONCE per scan
+ // (validToken() may refresh; legacy computes its storage map once in
doInitialize), threaded into
+ // the native-path URI normalization below so REST object-store reads
normalize via the vended
+ // credentials (a REST catalog's static storage map is empty by
design, so the static-only path
+ // would throw "No storage properties found for schema: oss"). Empty
for non-REST tables (FileIO
+ // gate in extractVendedToken) and offline unit tests (no context) →
the 2-arg normalize folds to
+ // the static-map path, leaving non-REST reads byte-unchanged.
+ Map<String, String> vendedToken =
+ context != null ? extractVendedToken(table) :
Collections.emptyMap();
+
+ // FIX-A1: the FE FileSplit proportional-weight denominator (legacy
PaimonScanNode:499, set on ALL
+ // splits). Session-only, so compute once here (before any split is
built). DISTINCT from the
+ // file-splitting targetSplitSize below — named weightDenominator to
make a positional swap impossible.
+ long weightDenominator = resolveSplitWeightDenominator(session);
+
// Non-DataSplit → always JNI
for (Split split : nonDataSplits) {
ranges.add(buildJniScanRange(split, tableLocation,
defaultFileFormat,
- Collections.emptyMap(), false));
+ Collections.emptyMap(), false, cppReader,
weightDenominator));
}
+ // COUNT(*) pushdown (FIX-COUNT-PUSHDOWN): collapse every split whose
merged (post-merge /
+ // post-deletion-vector) row count is precomputed into ONE count range
carrying the summed
+ // total, emitted after the loop — BE serves the count from
table_level_row_count (CountReader)
+ // without reading data. Mirrors legacy PaimonScanNode's count
short-circuit, which is the
+ // FIRST routing arm (BEFORE the native/JNI gate): a count-eligible
split must NOT also emit a
+ // data range, or BE would re-scan and double-count against deletion
vectors / PK merge. The
+ // collapse == legacy's <=10000 case (singletonList(first) +
assignCountToSplits([one], sum) ->
+ // one split bearing the full total); legacy's >10000 parallel-split
trim needs numBackends (an
+ // fe-core-only concern) and is intentionally dropped -> perf-only
divergence [deviations-log].
+ // Splits WITHOUT a precomputed merged count fall through to the
normal native/JNI routing so
+ // BE still counts them from file metadata / by reading.
+ long countSum = 0;
+ DataSplit countRepresentative = null;
+
+ // FIX-NATIVE-SUBSPLIT: target file split size for native ORC/Parquet
sub-splitting, computed
+ // lazily ONCE on the first native split (legacy
hasDeterminedTargetFileSplitSize parity).
+ long targetSplitSize = -1;
+
// Process DataSplits
for (DataSplit dataSplit : dataSplits) {
+ if (isCountPushdownSplit(countPushdown, dataSplit)) {
+ countSum += dataSplit.mergedRowCount();
+ if (countRepresentative == null) {
+ countRepresentative = dataSplit;
+ }
+ continue;
+ }
+
Map<String, String> partitionValues = getPartitionInfoMap(
- table, dataSplit.partition());
+ table, dataSplit.partition(), session.getTimeZone());
Optional<List<RawFile>> optRawFiles =
dataSplit.convertToRawFiles();
Optional<List<DeletionFile>> optDeletionFiles =
dataSplit.deletionFiles();
- if (supportNativeReader(optRawFiles)) {
- // Native reader path
+ if (shouldUseNativeReader(paimonHandle.isForceJni(),
Review Comment:
The `ignore_split_type` session variable (IGNORE_JNI / IGNORE_NATIVE debug
filtering) is not consumed anywhere on the plugin path — legacy honored it at
four points in PaimonScanNode.getSplits (master :368-369, :389, :431, :471),
and the variable still exists in SessionVariable at tip, so `SET
ignore_split_type=ignore_jni` now silently no-ops for paimon. Either wire it
into this routing loop or deprecate the variable; a silently dead debug knob is
the worst of both.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java:
##########
@@ -242,41 +640,311 @@ public Map<String, String> getScanNodeProperties(
props.put("paimon.options_json", sb.toString());
}
- // Location / storage properties
- for (Map.Entry<String, String> entry : properties.entrySet()) {
- String key = entry.getKey();
- if (key.startsWith("hadoop.") || key.startsWith("fs.")
- || key.startsWith("dfs.") || key.startsWith("hive.")
- || key.startsWith("s3.") || key.startsWith("cos.")
- || key.startsWith("oss.") || key.startsWith("obs.")) {
- props.put("location." + key, entry.getValue());
+ // FIX-STATIC-CREDS-BE (B-9): static catalog-level storage
credentials/config, normalized to
+ // BE-canonical keys (AWS_* for object stores). BE's native (FILE_S3)
reader understands ONLY the
+ // canonical keys, so the raw catalog aliases (s3.access_key,
oss.access_key, …) must be translated
+ // before they leave FE — copying them verbatim gives the native
reader no usable creds (403 on a
+ // private bucket). Sourced from the typed fe-filesystem
StorageProperties bound by fe-core and
+ // handed over via ctx.getStorageProperties() (P1-T04): each backend's
toBackendProperties().toMap()
+ // yields the canonical map (e.g. S3FileSystemProperties IS-A
BackendStorageProperties → AWS_*).
+ // This replaces the legacy getBackendStorageProperties() seam so the
connector derives BOTH its
+ // Hadoop config (P1-T03) and its BE creds from the SAME typed source
(design D-003). Empty when no
+ // context (offline unit tests) → no storage props emitted (never the
broken raw aliases).
+ //
+ // HDFS (DV-004 / R-007 — CLOSED by FU-T01): fe-filesystem now has a
typed HDFS BE model
+ // (HdfsFileSystemProperties); HdfsFileSystemProvider.bind() yields
it, so an HDFS-warehouse catalog
+ // emits the hadoop/dfs/HA/kerberos keys here (→ THdfsParams) at
parity with the legacy path
+ // (hadoop.config.resources resolved under the operator-configured
Config.hadoop_config_dir).
+ // KNOWN GAP 2 (R-008): the typed OSS/COS/OBS models omit
AWS_CREDENTIALS_PROVIDER_TYPE, which legacy
+ // emitted as ANONYMOUS for credential-less catalogs — a fe-filesystem
parity gap (out of P1 whitelist),
+ // tracked as a follow-up; only affects OSS/COS/OBS catalogs with no
static ak/sk.
+ if (context != null) {
+ Map<String, String> backendStorageProps = new HashMap<>();
+ for (StorageProperties sp : context.getStorageProperties()) {
+ sp.toBackendProperties().ifPresent(b ->
backendStorageProps.putAll(b.toMap()));
+ }
+ for (Map.Entry<String, String> e : backendStorageProps.entrySet())
{
+ props.put("location." + e.getKey(), e.getValue());
+ }
+ }
+
+ // FIX-REST-VENDED: overlay per-table vended cloud-storage credentials
(REST catalogs).
+ // The raw token is extracted from the live, snapshot-pinned table's
RESTTokenFileIO (paimon
+ // SDK only), then normalized to BE-facing AWS_* keys by the engine
(the connector cannot
+ // import fe-core's StorageProperties). Vended overlays static (legacy
precedence). Skipped
+ // when no context (offline unit tests) or the table is non-REST
(empty token -> no-op).
+ if (context != null) {
+ Map<String, String> vendedBeProps =
context.vendStorageCredentials(extractVendedToken(table));
+ for (Map.Entry<String, String> e : vendedBeProps.entrySet()) {
+ props.put("location." + e.getKey(), e.getValue());
+ }
+ }
+
+ // FIX-SCHEMA-EVOLUTION (B-1a): emit the native-reader schema
dictionary so BE matches file<->table
+ // columns BY FIELD ID across schema evolution (rename/reorder)
instead of falling back to NAME
+ // matching (which silently reads NULL/garbage for renamed columns).
Only meaningful when the table
+ // can take the native path: skip it when the handle name-forces JNI
(binlog/audit_log) OR the
+ // session forces JNI (force_jni_scanner) — in both cases every split
goes JNI and never consults
+ // the dict (FIX-FORCE-JNI-SCANNER: honor the same session escape
hatch the native router uses).
+ if (!paimonHandle.isForceJni() && !isForceJniScannerEnabled(session)) {
+ // The schema dict must be built from a FileStoreTable. A normal
data table IS one; a $ro
+ // (read-optimized) system table is a ReadOptimizedTable that
WRAPS a FileStoreTable and reads
+ // its data files with its field ids, so resolve the underlying
base FileStoreTable here.
+ Table schemaDictTable = resolveSchemaDictTable(table,
paimonHandle);
+ if (schemaDictTable != null) {
+ buildSchemaEvolutionParam(paimonHandle, schemaDictTable,
columns)
+ .ifPresent(v -> props.put(SCHEMA_EVOLUTION_PROP, v));
}
}
return props;
}
+ /**
+ * Resolves the {@link FileStoreTable} whose schema dictionary BE needs to
field-id-match the native
+ * data files for {@code table}. A normal data table IS the
FileStoreTable. A read-optimized system
+ * table ({@code $ro} → {@link ReadOptimizedTable}) is NOT a {@code
FileStoreTable} (it wraps one)
+ * but reads the BASE table's data files with the BASE field ids, so its
dict must come from the base
+ * FileStoreTable, reloaded here via the 2-arg base {@link Identifier}.
+ *
+ * <p>Restores legacy {@code PaimonScanNode} parity: legacy set {@code
history_schema_info} for ANY
+ * paimon table (incl. {@code $ro}) in {@code doInitialize}, so BE always
took the field-id path. The
+ * SPI connector had gated the dict on {@code instanceof FileStoreTable}
and so emitted nothing for
+ * {@code $ro}; with no {@code history_schema_info} BE's {@code
gen_table_info_node_by_field_id} fell
+ * into the legacy name-matching branch {@code
by_parquet_name(tuple_descriptor, ...)} and dereferenced
+ * a still-null tuple descriptor ({@code
table_schema_change_helper.cpp:94}) → a SIGSEGV that
+ * aborted the whole BE.
+ *
+ * <p>Returns {@code null} for a table with no native data files (metadata
system tables take the JNI
+ * path and never consult the dict), preserving the prior "emit nothing"
behavior for those.
+ */
+ private Table resolveSchemaDictTable(Table table, PaimonTableHandle
handle) {
+ if (table instanceof FileStoreTable) {
+ return table;
+ }
+ if (table instanceof ReadOptimizedTable) {
+ return reloadBaseTable(handle);
+ }
+ return null;
+ }
+
+ /**
+ * Reloads the BASE data table for a system handle via the 2-arg base
{@link Identifier}, under the
+ * FE-injected authenticator (D-052) when a context is present — mirroring
{@link #resolveTable}'s
+ * reload. Used to obtain the underlying {@link FileStoreTable} of a
{@code $ro} read so its schema
+ * dictionary can be emitted.
+ */
+ private Table reloadBaseTable(PaimonTableHandle handle) {
+ Identifier baseId = Identifier.create(handle.getDatabaseName(),
handle.getTableName());
+ try {
+ if (context == null) {
+ return catalogOps.getTable(baseId);
+ }
+ return context.executeAuthenticated(() ->
catalogOps.getTable(baseId));
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to load Paimon base table for
schema dict: " + baseId, e);
+ }
+ }
+
+ /**
+ * Extracts the raw per-table vended credential token from a REST catalog
table's
+ * {@link RESTTokenFileIO} (port of legacy {@code
PaimonVendedCredentialsProvider
+ * .extractRawVendedCredentials}, paimon SDK only). Returns empty for a
non-REST table (different
+ * FileIO) or when no valid token is available — the gate is the table's
FileIO type, equivalent
+ * to legacy's "metastore is REST" check for the read path.
+ */
+ static Map<String, String> extractVendedToken(Table table) {
+ if (table == null) {
+ return Collections.emptyMap();
+ }
+ FileIO fileIO = table.fileIO();
+ if (!(fileIO instanceof RESTTokenFileIO)) {
+ return Collections.emptyMap();
+ }
+ RESTToken token = ((RESTTokenFileIO) fileIO).validToken();
+ Map<String, String> raw = token == null ? null : token.token();
+ return raw == null ? Collections.emptyMap() : new HashMap<>(raw);
+ }
+
private PaimonScanRange buildJniScanRange(Split split, String
tableLocation,
String defaultFileFormat, Map<String, String> partitionValues,
- boolean isDataSplit) {
+ boolean isDataSplit, boolean cppReader, long weightDenominator) {
long splitWeight = 0;
if (isDataSplit) {
splitWeight = computeSplitWeight((DataSplit) split);
} else {
splitWeight = split.rowCount();
}
- String serializedSplit = encodeObjectToString(split);
+ String serializedSplit = encodeSplit(split, cppReader);
+ // FIX-JNI-FILE-FORMAT (P7-1): emit the real data-file format
(orc/parquet), NOT "jni". JNI routing
+ // is gated by the paimon.split property
(PaimonScanRange.populateRangeParams), so this string only
+ // feeds fileDesc.file_format, which BE's paimon_cpp_reader backfills
into FILE_FORMAT/MANIFEST_FORMAT
+ // (an invalid "jni" breaks the manifest read). Mirrors legacy
PaimonScanNode.setPaimonParams's
+ // fileDesc.setFileFormat(getFileFormat(...)).
return new PaimonScanRange.Builder()
- .fileFormat("jni")
+ .fileFormat(defaultFileFormat)
Review Comment:
`file_format` here comes from the table-level `file.format` option (default
`parquet`), while legacy derived it from the first data file's suffix with the
option only as fallback (master PaimonScanNode.java:259 + PaimonSplit.java:59)
— and the native path in this file still does suffix derivation. For a table
whose files are ORC but whose options omit `file.format` (or whose format
changed after writes), JNI/COUNT ranges now claim parquet. The only consumer
appears to be paimon_cpp_reader's FILE_FORMAT/MANIFEST_FORMAT backfill, so this
needs either an e2e check with an option-unset ORC table on the paimon-cpp
path, or simply restoring suffix derivation for these two builders to match the
native path.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateTableInfo.java:
##########
@@ -938,6 +933,8 @@ private static String
pluginCatalogTypeToEngine(PluginDrivenExternalCatalog cata
switch (catalog.getType()) {
case "max_compute":
return ENGINE_MAXCOMPUTE;
+ case "paimon":
Review Comment:
This is now the third fe-core string switch that must be hand-edited per
connector (with PluginDrivenExternalTable.getEngine() and
getEngineTableTypeName()); the javadoc itself says they 'must stay in sync'. P6
already added its iceberg cases to all three, confirming the N-connectors ×
N-edits trajectory. Connector identity (legacy engine name / table-type name)
looks like it belongs on the Connector SPI (e.g.
`Connector.getLegacyEngineName()`), declared once per plugin — otherwise every
future connector risks the exact silently-wrong-engine-name regression these
switches exist to prevent, with no compile-time guard tying them together.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonSchemaBuilder.java:
##########
@@ -0,0 +1,140 @@
+// 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.doris.connector.paimon;
+
+import org.apache.doris.connector.api.ConnectorColumn;
+import org.apache.doris.connector.api.DorisConnectorException;
+import org.apache.doris.connector.api.ddl.ConnectorCreateTableRequest;
+import org.apache.doris.connector.api.ddl.ConnectorPartitionField;
+import org.apache.doris.connector.api.ddl.ConnectorPartitionSpec;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.schema.Schema;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Builds a Paimon {@link Schema} from a connector-SPI
+ * {@link ConnectorCreateTableRequest}.
+ *
+ * <p>Functional port of the legacy fe-core
+ * {@code PaimonMetadataOps.toPaimonSchema}: primary keys come from
+ * {@code properties["primary-key"]}, partition keys come from the
+ * {@link ConnectorPartitionSpec} (identity transforms only), {@code
"primary-key"} and
+ * {@code "comment"} are stripped from the option map, and {@code "location"}
is re-keyed
+ * to {@link CoreOptions#PATH}. Bucket / distribution info is intentionally
NOT consumed —
+ * legacy paimon ignored bucketSpec and let any {@code bucket} option ride
through
+ * unchanged as a passthrough option.</p>
+ *
+ * <p>Two deliberate, safer divergences from the legacy bytes (each documented
+ tested at
+ * its call site): the table comment falls back to {@link
ConnectorCreateTableRequest#getComment()}
+ * (the {@code COMMENT} clause) when {@code properties["comment"]} is absent —
legacy read only
+ * the property and silently dropped the clause; and blank primary-key tokens
are filtered out —
+ * legacy would have forwarded an empty name that Paimon rejects
downstream.</p>
+ */
+public final class PaimonSchemaBuilder {
+
+ private static final String PRIMARY_KEY_IDENTIFIER = "primary-key";
+ private static final String PROP_COMMENT = "comment";
+ private static final String PROP_LOCATION = "location";
+ private static final String IDENTITY_TRANSFORM = "identity";
+
+ private PaimonSchemaBuilder() {
+ }
+
+ /**
+ * Convert a CREATE TABLE request into a Paimon {@link Schema}.
+ *
+ * @throws DorisConnectorException if a partition field uses a
non-identity transform
+ * or a column type cannot be represented in Paimon
+ */
+ public static Schema build(ConnectorCreateTableRequest request) {
+ Map<String, String> properties = request.getProperties();
+
+ // primary keys: from properties["primary-key"] only (no dedicated
request field),
+ // split on comma, trimmed, blanks dropped. Mirrors legacy
toPaimonSchema.
+ String pkAsString = properties.get(PRIMARY_KEY_IDENTIFIER);
+ List<String> primaryKeys = pkAsString == null
+ ? Collections.emptyList()
+ : Arrays.stream(pkAsString.split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .collect(Collectors.toList());
+
+ List<String> partitionKeys = partitionKeys(request.getPartitionSpec());
+
+ // options normalization: drop primary-key/comment, re-key location ->
CoreOptions.PATH.
+ Map<String, String> normalizedOptions = new HashMap<>(properties);
+ normalizedOptions.remove(PRIMARY_KEY_IDENTIFIER);
+ normalizedOptions.remove(PROP_COMMENT);
+ if (normalizedOptions.containsKey(PROP_LOCATION)) {
+ String path = normalizedOptions.remove(PROP_LOCATION);
+ normalizedOptions.put(CoreOptions.PATH.key(), path);
+ }
+
+ // comment resolution: legacy toPaimonSchema read ONLY
properties["comment"] (the nereids
+ // PROPERTIES("comment"=...) map); the dedicated COMMENT clause never
reached it. The SPI
+ // converter (CreateTableInfoToConnectorRequestConverter.convert) sets
request.getComment()
+ // from CreateTableInfo.getComment() (the COMMENT clause) and
request.getProperties() from
+ // CreateTableInfo.getProperties() (the PROPERTIES map) — two distinct
nereids fields.
+ // Resolution: properties["comment"] wins (preserves legacy
persisted-comment behavior),
+ // else fall back to request.getComment() so a user's COMMENT clause
is not silently dropped.
+ String comment = properties.containsKey(PROP_COMMENT)
+ ? properties.get(PROP_COMMENT)
+ : request.getComment();
+
+ Schema.Builder builder = Schema.newBuilder()
+ .options(normalizedOptions)
+ .primaryKey(primaryKeys)
+ .partitionKeys(partitionKeys)
+ .comment(comment);
+ for (ConnectorColumn col : request.getColumns()) {
+ // Column-level nullability applied here via copy(nullable),
mirroring legacy
+ // toPaimonSchema's
toPaimontype(type).copy(field.getContainsNull()).
+ builder.column(col.getName(),
+
PaimonTypeMapping.toPaimonType(col.getType()).copy(col.isNullable()),
+ col.getComment());
+ }
+ return builder.build();
+ }
+
+ private static List<String> partitionKeys(ConnectorPartitionSpec spec) {
+ if (spec == null) {
+ return Collections.emptyList();
+ }
+ List<String> keys = new ArrayList<>(spec.getFields().size());
+ for (ConnectorPartitionField field : spec.getFields()) {
Review Comment:
`spec.getInitialValues()` is never read, so `CREATE TABLE ... PARTITION BY
LIST (col) (PARTITION p VALUES IN ('x'))` silently accepts and drops the
explicit partition value definitions (verified live on this branch — table
created, definitions gone). Legacy also dropped them, so it's not a regression,
but the SPI now *models* the values (ConnectorPartitionValueDef /
getInitialValues, javadoc: 'only meaningful for LIST/RANGE') while the only
producer hardcodes `Collections.emptyList()` — dead API surface plus DDL
accepted that cannot be honored. Since this builder already fail-fasts on
unsupported transforms a few lines below, rejecting non-empty initial values
with a clear error would be consistent and honest.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonTypeMapping.java:
##########
@@ -177,6 +191,95 @@ private static ConnectorType toStructType(RowType rowType,
Options options) {
return ConnectorType.structOf(names, types);
}
+ /**
+ * Convert a Doris {@link ConnectorType} (as produced by the CREATE TABLE
request path)
+ * to a Paimon {@link DataType}.
+ *
+ * <p>This is the faithful reverse of the legacy fe-core
+ * {@code DorisToPaimonTypeVisitor}: the scalar set is intentionally
narrow (it mirrors
+ * the visitor's {@code atomic} branches and NOT MaxCompute's richer set),
CHAR/VARCHAR/STRING
+ * all collapse to {@code VarChar(MAX)} (declared length dropped),
DATETIME/DATETIMEV2 map to a
+ * plain {@code TimestampType()} (scale dropped), and the MAP key is
forced non-null. Types the
+ * legacy visitor did not handle (TINYINT, SMALLINT, LARGEINT, TIME,
IPV4/6, JSON, ...) throw,
+ * preserving the legacy gap.</p>
+ *
+ * <p>The returned type carries Paimon's default (nullable) flag;
column-level nullability is
+ * applied by the caller via {@code .copy(nullable)} (mirroring legacy
+ * {@code PaimonMetadataOps.toPaimonSchema}). The map-key {@code
.copy(false)} below is part of
+ * the type structure (not column nullability) and is kept.</p>
+ *
+ * @throws DorisConnectorException if the type cannot be represented in
Paimon
+ */
+ public static DataType toPaimonType(ConnectorType type) {
+ String name = type.getTypeName().toUpperCase(Locale.ROOT);
+ switch (name) {
+ case "BOOLEAN":
+ return new BooleanType();
+ case "INT":
+ case "INTEGER":
+ return new IntType();
+ case "BIGINT":
+ return new BigIntType();
+ case "FLOAT":
+ return new FloatType();
+ case "DOUBLE":
+ return new DoubleType();
+ case "CHAR":
+ case "VARCHAR":
+ case "STRING":
+ // Legacy parity: all char-family types collapse to
VarChar(MAX); declared
+ // length is intentionally dropped
(DorisToPaimonTypeVisitor.atomic isCharFamily).
+ return new VarCharType(VarCharType.MAX_LENGTH);
+ case "DATE":
+ case "DATEV2":
+ return new DateType();
+ case "DECIMALV2":
+ case "DECIMALV3":
+ case "DECIMAL32":
+ case "DECIMAL64":
+ case "DECIMAL128":
+ case "DECIMAL256":
+ return new DecimalType(type.getPrecision(), type.getScale());
+ case "DATETIME":
+ case "DATETIMEV2":
+ // Legacy parity: no-arg TimestampType (precision defaults to
6); the datetime
+ // scale is intentionally dropped to match
DorisToPaimonTypeVisitor.atomic, and it
+ // is a plain timestamp (NOT LocalZonedTimestampType).
+ return new TimestampType();
+ case "VARBINARY":
+ return new VarBinaryType(VarBinaryType.MAX_LENGTH);
+ case "VARIANT":
+ return new VariantType();
+ case "ARRAY":
+ return new ArrayType(toPaimonType(type.getChildren().get(0)));
Review Comment:
The to-Paimon (CREATE TABLE) direction drops nested nullability and
struct-field comments: ARRAY element / MAP value / ROW fields get default
nullability and no comment, while legacy DorisToPaimonTypeVisitor copied
`containsNull` per element/value/field and carried StructField comments (master
:57-78). Notably this is not an SPI limitation — ConnectorColumnConverter
carries childrenNullable + childrenComments and IcebergSchemaBuilder consumes
them; this connector just never reads them. Tables created through Doris now
get a different paimon schema than legacy produced (visible to Flink/Spark and
SHOW CREATE).
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
