github-actions[bot] commented on code in PR #66345:
URL: https://github.com/apache/doris/pull/66345#discussion_r3712979490
##########
fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorTableSchema.java:
##########
@@ -121,6 +124,18 @@ public ConnectorTableSchema(String tableName,
String tableFormatType,
Map<String, String> properties,
Set<ConnectorCapability> tableCapabilities) {
+ this(tableName, columns, tableFormatType, properties,
tableCapabilities, null);
Review Comment:
[P1] Preserve identities when rebuilding delegated schemas.
`HiveConnectorMetadata.reflectSiblingCapabilities()` reconstructs every
Iceberg-on-HMS schema through this five-argument overload, so the identity
produced by `IcebergConnectorMetadata` is reset to null. That null reaches the
bound write handle and makes both provider- and transaction-side Iceberg fences
skip, disabling the UUID/schema/spec/sort/mode protections for the delegated
path. Please pass `siblingSchema.getWriteMetadataIdentity()` at that copy site
and cover an Iceberg-on-HMS write.
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveWritePlanProvider.java:
##########
@@ -99,19 +100,75 @@ public ConnectorSinkPlan planWrite(ConnectorSession
session, ConnectorWriteHandl
HiveTableHandle tableHandle = (HiveTableHandle)
handle.getTableHandle();
HiveConnectorTransaction transaction = currentTransaction(session);
- // Load the table under the catalog auth context; it drives both the
location resolution
- // (buildWriteContext) and the sink assembly (buildSink). beginWrite
re-loads it for its own
- // begin-guard — the double-load is accepted (mirrors iceberg),
keeping the flow simple.
+ // One fresh table generation drives validation, transaction state,
location resolution, and sink
+ // assembly. Reloading in beginWrite would reopen a TOCTOU window
after the schema fence.
HmsTableInfo table = loadTable(tableHandle);
+ validateBoundWriteMetadata(table, handle);
HiveWriteContext writeContext = buildWriteContext(session,
tableHandle, table, handle);
- transaction.beginWrite(session, tableHandle.getDbName(),
tableHandle.getTableName(), writeContext);
+ transaction.beginWrite(session, tableHandle.getDbName(),
tableHandle.getTableName(), writeContext, table);
THiveTableSink sink = buildSink(session, tableHandle, table, handle,
writeContext);
TDataSink dataSink = new TDataSink(TDataSinkType.HIVE_TABLE_SINK);
dataSink.setHiveTableSink(sink);
return new ConnectorSinkPlan(dataSink);
}
+ private static void validateBoundWriteMetadata(HmsTableInfo table,
ConnectorWriteHandle handle) {
+ String boundIdentity = handle.getBoundWriteMetadataIdentity();
+ if (boundIdentity != null &&
!boundIdentity.equals(writeMetadataIdentity(table))) {
+ // Bound expressions and live THiveTableSink ordinals must
describe one HMS schema generation;
+ // accepting a reorder here silently writes each value under
another column name.
+ throw new DorisConnectorException(
+ "Hive write metadata changed after the write was bound;
retry the statement");
+ }
+
+ List<ConnectorColumn> boundColumns = handle.getBoundTargetColumns();
+ int liveColumnCount = table.getColumns().size() +
table.getPartitionKeys().size();
+ if (boundColumns.isEmpty()) {
+ return;
+ }
+ if (boundColumns.size() != liveColumnCount) {
+ throw schemaChangedException();
+ }
+ for (int i = 0; i < boundColumns.size(); i++) {
+ ConnectorColumn live = i < table.getColumns().size()
+ ? table.getColumns().get(i)
+ : table.getPartitionKeys().get(i -
table.getColumns().size());
+ if
(!boundColumns.get(i).getName().equalsIgnoreCase(live.getName())) {
+ throw schemaChangedException();
+ }
+ }
+ }
+
+ private static DorisConnectorException schemaChangedException() {
+ return new DorisConnectorException(
+ "Hive table schema changed after the write was bound; retry
the statement");
+ }
+
+ static String writeMetadataIdentity(HmsTableInfo table) {
+ StringBuilder identity = new StringBuilder();
+ appendMetadataToken(identity, "data-columns");
+ for (ConnectorColumn column : table.getColumns()) {
+ appendColumnIdentity(identity, column);
+ }
+ appendMetadataToken(identity, "partition-columns");
+ for (ConnectorColumn column : table.getPartitionKeys()) {
+ appendColumnIdentity(identity, column);
+ }
+ return identity.toString();
+ }
+
+ private static void appendColumnIdentity(StringBuilder identity,
ConnectorColumn column) {
+ appendMetadataToken(identity,
column.getName().toLowerCase(Locale.ROOT));
+ appendMetadataToken(identity, column.getType());
Review Comment:
[P1] Fence the exact effective Hive schema, not this raw top-level
rendering. `ConnectorType.toString()` collapses every ARRAY/MAP/STRUCT to its
outer tag, so nested field changes collide; `getTableSchema` can also turn raw
OpenCSV columns into STRING based on the SerDe and enrich defaults from a
separate RPC, neither of which is represented here. The count/name fallback
misses all three, letting BE serialize S0's nested layout, string expression,
or materialized default under S1 metadata. Please build and compare a recursive
identity from the same effective columns/defaults used by binding (including
the fresh SerDe/default inputs), with nested-type, OpenCSV-to-LazySimple, and
default-only race tests.
##########
fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/write/ConnectorWritePlanProvider.java:
##########
@@ -98,6 +98,25 @@ default List<ConnectorWriteSortColumn>
getWriteSortColumns(ConnectorSession sess
return null;
}
+ /**
+ * Resolves write-sort positions against the bind-time target schema.
Connectors with stable field
+ * identities should override this form; the default preserves existing
name/ordinal behavior.
+ */
+ default List<ConnectorWriteSortColumn>
getWriteSortColumns(ConnectorSession session,
+ ConnectorTableHandle tableHandle, List<ConnectorColumn>
boundTargetColumns) {
+ return getWriteSortColumns(session, tableHandle);
+ }
+
+ /**
+ * Returns an opaque identity for metadata that shapes the physical write
plan. The engine captures it
+ * while binding the sink and returns it through {@link
ConnectorWriteHandle}; connectors can reject the
+ * write if a later metadata refresh would make that physical plan stale.
Default: {@code null} when the
+ * connector has no such metadata fence.
+ */
+ default String getWriteMetadataIdentity(ConnectorSession session,
ConnectorTableHandle tableHandle) {
Review Comment:
[P1] Bump the connector plugin API major for this SPI addition. This method
(together with the new write-handle/schema methods in this PR) is public plugin
surface, but `fe/fe-connector/pom.xml` remains at 3.0 and the surface artifacts
are unchanged. The loader compares majors only, under the explicit rule that
even additions bump the major; otherwise a plugin built here can be admitted by
an older major-3 FE and fail with `NoSuchMethodError` when it calls a new
handle/provider/schema method. Please regenerate the required baselines and
bump the connector API major in this commit.
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HiveWritePlanProvider.java:
##########
@@ -99,19 +100,75 @@ public ConnectorSinkPlan planWrite(ConnectorSession
session, ConnectorWriteHandl
HiveTableHandle tableHandle = (HiveTableHandle)
handle.getTableHandle();
HiveConnectorTransaction transaction = currentTransaction(session);
- // Load the table under the catalog auth context; it drives both the
location resolution
- // (buildWriteContext) and the sink assembly (buildSink). beginWrite
re-loads it for its own
- // begin-guard — the double-load is accepted (mirrors iceberg),
keeping the flow simple.
+ // One fresh table generation drives validation, transaction state,
location resolution, and sink
+ // assembly. Reloading in beginWrite would reopen a TOCTOU window
after the schema fence.
HmsTableInfo table = loadTable(tableHandle);
+ validateBoundWriteMetadata(table, handle);
Review Comment:
[P1] Keep the Hive table generation protected through publication. This is
the only generation check, but the identity has no table-incarnation token, so
a same-shaped drop/recreate already passes here; after this call,
`HiveConnectorTransaction.commit()` never reloads or validates the table. A
replacement during BE execution can make the committer move S0 files to the
captured old location while statistics/partition mutations address the current
object by db/name, or let a reordered schema reinterpret the S0 positional
layout. The Iceberg UUID/OCC path does not protect Hive. Please carry a stable
HMS incarnation from the bind-time load and protect or atomically revalidate it
at commit before publishing, with same-shaped replacement and post-plan DDL
races.
--
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]
