github-actions[bot] commented on code in PR #67904:
URL: https://github.com/apache/doris/pull/67904#discussion_r3999084520
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java:
##########
@@ -196,8 +197,12 @@ public static FileStoreTable applyOptionsWithoutTimeTravel(
.filter(key -> !tableOptions.containsKey(key))
.forEach(key -> isolatedOptions.put(key, null));
}
- // The statement fence already selected the schema generation.
Preserve that generation
- // while carrying only the resolved read selector and execution
options into this copy.
+ String schemaId = options.get(BOUND_SCHEMA_ID);
+ if (schemaId != null && table.schema().id() !=
Long.parseLong(schemaId)) {
+ // A cached table can predate binding, and latest can advance
again after binding.
+ // Copy the exact schema while retaining catalog options,
decorators and branch identity.
Review Comment:
[P1] Keep schema-owned options from the same generation as these fields.
This copies the bound schema S2 with `table.options()` from the currently
loaded table, which can be stale S1 or later S3. Paimon rewrites
field-referencing options such as `bucket-key`/`sequence.field` during renames;
for example, restoring S2 fields after a key rename with S1's bucket-key name
fails `TableSchema` validation or changes read semantics. Preserve S2's
schema-owned options and overlay only the catalog/runtime options that must
survive this copy; add a warm-cache rename test for a bucket/sequence field.
##########
regression-test/suites/external_table_p0/paimon/test_paimon_schema_only_snapshot_precision.groovy:
##########
@@ -0,0 +1,125 @@
+// 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.
+
+suite("test_paimon_schema_only_snapshot_precision", "p0,external,paimon") {
+ String enabled = context.config.otherConfigs.get("enablePaimonTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable paimon test")
+ return
+ }
+
+ String minioPort = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String catalogName = "test_paimon_schema_only_snapshot_precision"
+ String dbName = "paimon_schema_only_snapshot_precision_db"
+ String tableName = "schema_only_timeline"
+ String branchName = "schema_only_branch"
+
+ def latestSnapshotId = {
+ List<List<Object>> rows = spark_paimon """
+ select snapshot_id
+ from paimon.${dbName}.`${tableName}\$snapshots`
+ order by snapshot_id desc
+ limit 1
+ """
+ assertEquals(1, rows.size())
+ return rows[0][0].toString()
+ }
+
+ sql """drop catalog if exists ${catalogName}"""
+ sql """
+ create catalog ${catalogName} properties (
+ 'type'='paimon',
+ 'warehouse'='s3://warehouse/wh',
+ 's3.endpoint'='http://${externalEnvIp}:${minioPort}',
+ 's3.access_key'='admin',
+ 's3.secret_key'='password',
+ 's3.path.style.access'='true',
+ 'meta.cache.paimon.table.ttl-second'='0'
+ )
+ """
+
+ try {
+ // Explicit NTZ makes this exercise predicate pushdown instead of LTZ
residual filtering.
+ spark_paimon_multi """
+ create database if not exists paimon.${dbName};
+ drop table if exists paimon.${dbName}.${tableName};
+ create table paimon.${dbName}.${tableName} (
+ id int,
+ old_name string,
+ event_time timestamp_ntz
+ ) using paimon
+ tblproperties ('file.format'='parquet');
+ insert into paimon.${dbName}.${tableName}
+ values (1, 'base', timestamp_ntz '2024-01-01 00:00:00.123456');
+ """
+ String dataSnapshotId = latestSnapshotId()
+ spark_paimon_multi """
+ call paimon.sys.create_tag(
+ table => '${dbName}.${tableName}',
+ tag => 'schema_base'
+ );
+ call paimon.sys.create_branch(
+ '${dbName}.${tableName}',
+ '${branchName}',
+ 'schema_base'
+ );
+ alter table paimon.${dbName}.`${tableName}\$branch_${branchName}`
+ rename column old_name to branch_name;
+ alter table paimon.${dbName}.${tableName}
+ rename column old_name to current_name;
+ """
+
+ // A schema-only rename must leave the data snapshot unchanged;
otherwise these queries
+ // would not exercise the split between current schema binding and
snapshot-pinned data.
+ assertEquals(dataSnapshotId, latestSnapshotId())
+
+ sql """switch ${catalogName}"""
+ sql """use ${dbName}"""
+ sql """refresh table ${tableName}"""
+
+ assertEquals([[1, "base"]], sql("""
Review Comment:
[P2] Record these deterministic rows through the regression harness. The
five fixed result checks below use `assertEquals(sql(...))`, so this new suite
has no generated `.out` baseline even though root `AGENTS.md` requires
determined results to use `qt_`/`order_qt_` cases. Please keep the snapshot-id
setup assertion, but convert the stable plain, OPTIONS, branch, native, and JNI
queries to uniquely named `order_qt_` cases and generate the matching output
file.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonPredicateConverter.java:
##########
@@ -329,15 +328,10 @@ private Object convertLiteralValue(ConnectorLiteral
literal, DataType paimonType
}
return null;
case TIMESTAMP_WITHOUT_TIME_ZONE:
- // Zone-free type: interpret the literal's wall-clock in UTC
to match paimon's
- // stored min/max file/partition stats (computed by reading
the wall clock as UTC).
- // Mirrors legacy PaimonValueConverter#visit(TimestampType),
which uses a fixed
- // GMT Calendar. Using the session zone here would shift the
epoch-millis vs the
- // stored stats and risk false file/partition pruning = silent
data loss.
+ // Preserve the complete wall-clock value: narrowing it to
epoch milliseconds can
+ // make Paimon prune every file matching a
non-millisecond-aligned predicate.
if (value instanceof LocalDateTime) {
- LocalDateTime dt = (LocalDateTime) value;
- long millis = dt.toInstant(ZoneOffset.UTC).toEpochMilli();
- return Timestamp.fromEpochMillis(millis);
Review Comment:
[P1] Avoid exact pushdown when source timestamp precision exceeds Doris's
scale. Paimon `TIMESTAMP(7..9)` is exposed as `DATETIMEV2(6)`, and the readers
truncate sub-microsecond digits. A stored `.123456789` therefore compares equal
to `.123456` in Doris, but this creates an exact Paimon literal `.123456000`;
equality/file statistics can prune the row before residual evaluation. Gate
pushdown for source precision above 6 or translate comparisons into safe
microsecond ranges, and add a `TIMESTAMP(9)` test with nonzero sub-microsecond
digits.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanParams.java:
##########
@@ -196,8 +197,12 @@ public static FileStoreTable applyOptionsWithoutTimeTravel(
.filter(key -> !tableOptions.containsKey(key))
.forEach(key -> isolatedOptions.put(key, null));
}
- // The statement fence already selected the schema generation.
Preserve that generation
- // while carrying only the resolved read selector and execution
options into this copy.
+ String schemaId = options.get(BOUND_SCHEMA_ID);
+ if (schemaId != null && table.schema().id() !=
Long.parseLong(schemaId)) {
+ // A cached table can predate binding, and latest can advance
again after binding.
+ // Copy the exact schema while retaining catalog options,
decorators and branch identity.
Review Comment:
[P1] Rebuild fallback branches separately here. When `table` is a
`FallbackReadFileStoreTable`, `copy(TableSchema)` applies this main schema to
both children; Paimon 1.3.1 only preserves the fallback branch option if the
incoming schema already contains `branch`. A normal main schema does not, so
the fallback child defaults back to main, both scans read main, and
fallback-only partitions can be silently omitted. Preserve each child's
branch/schema provenance (and privilege wrapper) rather than sending one schema
through polymorphic copy, and cover a stale privilege-decorated fallback pair.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/mvcc/PluginDrivenMvccExternalTable.java:
##########
@@ -172,6 +172,14 @@ private PluginDrivenMvccSnapshot materializeLatest(
// legacy listPartitions/LIST/timestamp path below (byte-unchanged;
the no-op applySnapshot for the
// latest pin is side-effect-free for both paimon and iceberg).
ConnectorTableHandle pinnedHandle = metadata.applySnapshot(session,
handle, connectorSnapshot);
+ PluginDrivenSchemaCacheValue pinnedSchema = null;
+ if (connectorSnapshot.getSchemaId() >= 0) {
+ // Latest data and schema can advance independently. Keep the
connector's exact schema
+ // on the statement pin so analysis cannot fall back to a
different cached generation.
+ ConnectorTableSchema atSchema = metadata.getTableSchema(session,
pinnedHandle, connectorSnapshot);
Review Comment:
[P1] Keep connector-generic pinned schemas internally version-consistent.
Iceberg's `getTableSchema(..., snapshot A)` takes columns from schema A, but
its shared builder still derives `partition_columns` from the live current
table/schema. With REST vended credentials the latest pin is cached while the
table is reloaded each query, so a schema-only partition-source rename can
yield historical column `old` plus `partition_columns=new`;
`toSchemaCacheValue` then silently drops the unmatched partition column. Please
make Iceberg derive partition identity from the same historical schema/spec
coordinate before installing this generic pin, and add a
latest-cache-hit/live-table rename test.
--
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]
