This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 8a4868a9da0 [fix](fe) Normalize connector table errors (#66628)
8a4868a9da0 is described below
commit 8a4868a9da0e2392acdd00bbfd7f37fc20a5e325
Author: Gabriel <[email protected]>
AuthorDate: Thu Aug 13 14:06:16 2026 +0800
[fix](fe) Normalize connector table errors (#66628)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: Several negative connector queries expose generic
plugin or filesystem implementation details instead of stable
connector-facing errors. Equivalent failures can also produce different
messages during metadata binding, system-table resolution, and scan
planning. Use the connector display name for system-table constraint
errors, report a normal table-not-found error when a remotely resolved
handle disappears, and normalize Iceberg metadata-file misses across all
table-loading entry points by inspecting the complete exception cause
chain.
### Release note
Paimon and Iceberg negative queries now return stable connector-facing
errors for unsupported system-table scans, missing redirected tables,
and missing Iceberg metadata files.
### Check List (For Author)
- Test
- [ ] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason
- Behavior changed:
- [ ] No.
- [x] Yes. Connector errors no longer expose generic plugin or
filesystem wrapper details in these paths.
- Does this need documentation?
- [x] No.
- [ ] Yes.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
.../iceberg/IcebergConnectorMetadata.java | 28 ++++--
.../connector/iceberg/IcebergExceptionUtils.java | 61 +++++++++++++
.../connector/iceberg/IcebergScanPlanProvider.java | 96 +++++++++++++-------
.../iceberg/IcebergConnectorMetadataMvccTest.java | 32 +++++++
.../iceberg/IcebergConnectorMetadataTest.java | 50 +++++++++++
.../iceberg/IcebergScanPlanProviderTest.java | 100 +++++++++++++++++++++
.../iceberg/RecordingIcebergCatalogOps.java | 10 +++
.../datasource/scan/PluginDrivenScanNode.java | 29 ++++--
.../scan/PluginDrivenScanNodeSysHandleTest.java | 53 ++++++++++-
.../PluginDrivenScanNodeSysTableGuardTest.java | 81 +++++++++++++++++
10 files changed, 490 insertions(+), 50 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
index 0e6425a8f07..2546fe4690e 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadata.java
@@ -412,7 +412,15 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
try {
exists = context.executeAuthenticated(() ->
catalogOps.tableExists(dbName, tableName));
} catch (Exception e) {
- throw new RuntimeException("Failed to check table exist, error
message is:" + e.getMessage(), e);
+ // Preserve Optional.empty even when an auth/catalog layer wraps
NoSuchTableException.
+ if (ExceptionUtils.getThrowableList(e).stream()
+ .anyMatch(NoSuchTableException.class::isInstance)) {
+ return Optional.empty();
+ }
+ // Existence checks in several catalogs load metadata internally,
so normalize them at the shared
+ // handle boundary just like explicit table loads.
+ throw IcebergExceptionUtils.wrapTableLoadFailure(new
IcebergTableHandle(dbName, tableName), e,
+ "Failed to check table exist, error message is:");
}
if (!exists) {
return Optional.empty();
@@ -646,7 +654,8 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
try {
return context.executeAuthenticated(() ->
resolveTableForRead(session, handle));
} catch (Exception e) {
- throw new RuntimeException("Failed to load table, error message
is:" + e.getMessage(), e);
+ throw IcebergExceptionUtils.wrapTableLoadFailure(
+ handle, e, "Failed to load table, error message is:");
}
}
@@ -688,7 +697,8 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
base,
MetadataTableType.from(handle.getSysTableName()));
});
} catch (Exception e) {
- throw new RuntimeException("Failed to load table, error message
is:" + e.getMessage(), e);
+ throw IcebergExceptionUtils.wrapTableLoadFailure(
+ handle, e, "Failed to load table, error message is:");
}
}
@@ -1919,8 +1929,8 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
() -> buildMvccPartitionViewUncached(session,
iceHandle)));
});
} catch (Exception e) {
- throw new RuntimeException("Failed to build iceberg MVCC partition
view, error message is:"
- + e.getMessage(), e);
+ throw IcebergExceptionUtils.wrapTableLoadFailure(iceHandle, e,
+ "Failed to build iceberg MVCC partition view, error
message is:");
}
}
@@ -1967,8 +1977,8 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
TableIdentifier.of(iceHandle.getDbName(),
iceHandle.getTableName()), partitionCache);
});
} catch (Exception e) {
- throw new RuntimeException("Failed to list iceberg partition
names, error message is:"
- + e.getMessage(), e);
+ throw IcebergExceptionUtils.wrapTableLoadFailure(iceHandle, e,
+ "Failed to list iceberg partition names, error message
is:");
}
}
@@ -2004,8 +2014,8 @@ public class IcebergConnectorMetadata implements
ConnectorMetadata {
return listPartitionsViewCache.get(key, () ->
listPartitionsUncached(session, iceHandle));
});
} catch (Exception e) {
- throw new RuntimeException("Failed to list iceberg partitions,
error message is:"
- + e.getMessage(), e);
+ throw IcebergExceptionUtils.wrapTableLoadFailure(iceHandle, e,
+ "Failed to list iceberg partitions, error message is:");
}
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergExceptionUtils.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergExceptionUtils.java
new file mode 100644
index 00000000000..db84ade0d7f
--- /dev/null
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergExceptionUtils.java
@@ -0,0 +1,61 @@
+// 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.iceberg;
+
+import org.apache.doris.connector.spi.DorisConnectorException;
+
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.iceberg.exceptions.NotFoundException;
+
+import java.io.FileNotFoundException;
+
+final class IcebergExceptionUtils {
+
+ private IcebergExceptionUtils() {
+ }
+
+ static RuntimeException wrapTableLoadFailure(
+ IcebergTableHandle handle, Exception failure, String
fallbackPrefix) {
+ if (isMetadataNotFound(failure)) {
+ return metadataNotFound(handle, failure);
+ }
+ return new RuntimeException(fallbackPrefix + failure.getMessage(),
failure);
+ }
+
+ static RuntimeException wrapMetadataReadFailure(
+ IcebergTableHandle handle, RuntimeException failure) {
+ if (isMetadataNotFound(failure)) {
+ return metadataNotFound(handle, failure);
+ }
+ return failure;
+ }
+
+ private static boolean isMetadataNotFound(Throwable failure) {
+ // Iceberg and FileIO implementations wrap missing metadata
differently. Inspect every cause so eager
+ // planners and background lazy split iteration preserve one stable
table-scoped error contract.
+ return ExceptionUtils.getThrowableList(failure).stream()
+ .anyMatch(cause -> cause instanceof NotFoundException
+ || cause instanceof FileNotFoundException);
+ }
+
+ private static DorisConnectorException metadataNotFound(
+ IcebergTableHandle handle, Throwable failure) {
+ return new DorisConnectorException("Metadata not found in metadata
location for table "
+ + handle.getDbName() + "." + handle.getTableName(), failure);
+ }
+}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
index 4f0f8417c66..02c5ac5bcf8 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java
@@ -418,8 +418,18 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
*/
@Override
public List<ConnectorScanRange> planScan(ConnectorSession session,
ConnectorScanRequest request) {
- return planScanInternal(session, request.getTableHandle(),
request.getColumns(),
- request.getFilter(), request.isCountPushdown());
+ IcebergTableHandle handle = (IcebergTableHandle)
request.getTableHandle();
+ try {
+ return planScanInternal(session, handle, request.getColumns(),
+ request.getFilter(), request.isCountPushdown());
+ } catch (RuntimeException e) {
+ // Normal data scans and native position_deletes run on File
Scanner V2. Keep the serialized JNI
+ // system-table route untouched because its deferred reads belong
to the V1 scanner contract.
+ if (!handle.isSystemTable() || isPositionDeletesSysTable(handle)) {
+ throw IcebergExceptionUtils.wrapMetadataReadFailure(handle, e);
+ }
+ throw e;
+ }
}
/**
@@ -465,8 +475,10 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
fileCount += (added == null ? 0 : added) + (existing == null ?
0 : existing);
}
} catch (IOException e) {
- throw new RuntimeException("Failed to count iceberg manifest files
for batch decision, error message is:"
- + e.getMessage(), e);
+ throw IcebergExceptionUtils.wrapTableLoadFailure(iceHandle, e,
+ "Failed to count iceberg manifest files for batch
decision, error message is:");
+ } catch (RuntimeException e) {
+ throw IcebergExceptionUtils.wrapMetadataReadFailure(iceHandle, e);
}
return fileCount >= threshold ? fileCount : -1;
}
@@ -501,9 +513,15 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
long fileSplitSize = sessionLong(session, FILE_SPLIT_SIZE, 0L);
long sliceSize = fileSplitSize > 0 ? fileSplitSize
: sessionLong(session, MAX_FILE_SPLIT_SIZE,
DEFAULT_MAX_FILE_SPLIT_SIZE);
- CloseableIterable<FileScanTask> tasks = streamingFileScanTasks(scan,
session, table, filter, sliceSize);
- return new IcebergStreamingSplitSource(tasks, table, formatVersion,
partitioned,
- orderedPartitionKeys, zone, uriNormalizer, sliceSize,
iceHandle.getRewriteFileScope());
+ try {
+ CloseableIterable<FileScanTask> tasks = streamingFileScanTasks(
+ scan, session, table, filter, sliceSize);
+ return new IcebergStreamingSplitSource(tasks, table,
formatVersion, partitioned,
+ orderedPartitionKeys, zone, uriNormalizer, sliceSize,
+ iceHandle.getRewriteFileScope(), iceHandle);
+ } catch (RuntimeException e) {
+ throw IcebergExceptionUtils.wrapMetadataReadFailure(iceHandle, e);
+ }
}
private static ConnectorSplitSource emptySplitSource() {
@@ -564,6 +582,7 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
private final UnaryOperator<String> uriNormalizer;
private final long sliceSize;
private final Set<String> rewriteScope;
+ private final IcebergTableHandle handle;
// Lazily opened on first hasNext() so the ctor never throws —
iceberg's ParallelIterable submits
// manifest readers in tasks.iterator(), which can fail; opening it
eagerly here would throw out of
// streamSplits() BEFORE the source is returned, leaking the
planFiles() iterable (the engine pump's
@@ -578,7 +597,8 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
IcebergStreamingSplitSource(CloseableIterable<FileScanTask> tasks,
Table table, int formatVersion,
boolean partitioned, List<String> orderedPartitionKeys, ZoneId
zone,
- UnaryOperator<String> uriNormalizer, long sliceSize,
Set<String> rewriteScope) {
+ UnaryOperator<String> uriNormalizer, long sliceSize,
Set<String> rewriteScope,
+ IcebergTableHandle handle) {
this.tasks = tasks;
this.table = table;
this.formatVersion = formatVersion;
@@ -588,25 +608,31 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
this.uriNormalizer = uriNormalizer;
this.sliceSize = sliceSize;
this.rewriteScope = rewriteScope;
+ this.handle = handle;
}
@Override
public boolean hasNext() {
- if (buffered != null) {
- return true;
- }
- if (iterator == null) {
- iterator = tasks.iterator();
- }
- while (iterator.hasNext()) {
- IcebergScanRange range = buildRangeForTask(iterator.next(),
table, formatVersion, partitioned,
- orderedPartitionKeys, zone, uriNormalizer, sliceSize,
rewriteScope, null, scratch);
- if (range != null) {
- buffered = range;
+ try {
+ if (buffered != null) {
return true;
}
+ if (iterator == null) {
+ iterator = tasks.iterator();
+ }
+ while (iterator.hasNext()) {
+ IcebergScanRange range =
buildRangeForTask(iterator.next(), table, formatVersion, partitioned,
+ orderedPartitionKeys, zone, uriNormalizer,
sliceSize, rewriteScope, null, scratch);
+ if (range != null) {
+ buffered = range;
+ return true;
+ }
+ }
+ return false;
+ } catch (RuntimeException e) {
+ // Lazy manifest opening happens on the split-pump thread,
after streamSplits has returned.
+ throw IcebergExceptionUtils.wrapMetadataReadFailure(handle, e);
}
- return false;
}
@Override
@@ -1606,7 +1632,11 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
if (!systemTable) {
scanSchema = pinnedSchema(table, iceHandle);
exactScan = buildScan(table, iceHandle, filter, session);
- applicableEqualityDeleteFieldIds =
cachedApplicableEqualityDeleteFieldIds(table, exactScan);
+ try {
+ applicableEqualityDeleteFieldIds =
cachedApplicableEqualityDeleteFieldIds(table, exactScan);
+ } catch (RuntimeException e) {
+ throw IcebergExceptionUtils.wrapMetadataReadFailure(iceHandle,
e);
+ }
hasApplicableEqualityDeletes =
!applicableEqualityDeleteFieldIds.isEmpty();
Optional<Map<Integer, List<String>>> nameMapping =
IcebergSchemaUtils.extractNameMapping(table);
if (requiresCurrentScanSemantics(
@@ -1625,11 +1655,17 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
// PERF-03: the non-system format resolution falls back to an
unfiltered whole-table planFiles() when the
// table sets neither write-format nor write.format.default; memoize
that inference per (table, snapshot)
// across queries via formatCache (pure metadata, no credential gate).
Null cache (offline) resolves live.
- props.put(ScanNodePropertyKeys.FILE_FORMAT_TYPE,
- systemTable ? "jni"
- : IcebergWriterHelper.getFileFormat(table,
+ String fileFormatType = "jni";
+ if (!systemTable) {
+ try {
+ fileFormatType = IcebergWriterHelper.getFileFormat(table,
TableIdentifier.of(iceHandle.getDbName(),
iceHandle.getTableName()), formatCache)
- .name().toLowerCase(Locale.ROOT));
+ .name().toLowerCase(Locale.ROOT);
+ } catch (RuntimeException e) {
+ throw IcebergExceptionUtils.wrapMetadataReadFailure(iceHandle,
e);
+ }
+ }
+ props.put(ScanNodePropertyKeys.FILE_FORMAT_TYPE, fileFormatType);
// [D-065] System (metadata) tables ($snapshots/$files/...) read via
the JNI serialized-split path
// (planSystemTableScan): the metadata-table schema travels INSIDE the
serialized FileScanTask, so BE
// needs neither the base-table path_partition_keys (a metadata table
is not base-spec partitioned ->
@@ -2801,13 +2837,13 @@ public class IcebergScanPlanProvider implements
ConnectorScanPlanProvider {
// re-validates the credential even on a scope hit).
IcebergCatalogOps ops = catalogOpsResolver.apply(session);
Table raw = IcebergStatementScope.sharedTable(session,
handle.getDbName(), handle.getTableName(), () -> {
- if (context == null) {
- return loadRawTable(ops, handle);
- }
try {
- return context.executeAuthenticated(() -> loadRawTable(ops,
handle));
+ return context == null
+ ? loadRawTable(ops, handle)
+ : context.executeAuthenticated(() -> loadRawTable(ops,
handle));
} catch (Exception e) {
- throw new RuntimeException("Failed to load table for scan,
error message is:" + e.getMessage(), e);
+ throw IcebergExceptionUtils.wrapTableLoadFailure(
+ handle, e, "Failed to load table for scan, error
message is:");
}
});
return wrapTableForScan(raw);
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataMvccTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataMvccTest.java
index be8514b11fa..6aec247222b 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataMvccTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataMvccTest.java
@@ -34,6 +34,7 @@ import org.apache.iceberg.Schema;
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NotFoundException;
import org.apache.iceberg.inmemory.InMemoryCatalog;
import org.apache.iceberg.types.Types;
import org.junit.jupiter.api.Assertions;
@@ -536,4 +537,35 @@ public class IcebergConnectorMetadataMvccTest {
Assertions.assertEquals(1, ctx.authCount);
Assertions.assertFalse(ops.log.contains("loadTable:db1.t1"),
"loadTable must sit inside executeAuthenticated");
}
+
+ @Test
+ public void listPartitionNamesNormalizesDeepMetadataNotFoundFailure() {
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ ops.loadTableFailure = new RuntimeException("catalog wrapper",
+ new RuntimeException("storage wrapper", new
NotFoundException("metadata is missing")));
+ IcebergConnectorMetadata md = new IcebergConnectorMetadata(
+ ops, IcebergCatalogProperties.of(Collections.emptyMap()), new
RecordingConnectorContext());
+
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> md.listPartitionNames(null, handle()));
+ Assertions.assertEquals("Metadata not found in metadata location for
table db1.t1", ex.getMessage());
+ }
+
+ @Test
+ public void parallelPartitionReadersNormalizeDeepMetadataNotFoundFailure()
{
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ ops.loadTableFailure = new RuntimeException("catalog wrapper",
+ new RuntimeException("storage wrapper", new
NotFoundException("metadata is missing")));
+ IcebergConnectorMetadata md = new IcebergConnectorMetadata(
+ ops, IcebergCatalogProperties.of(Collections.emptyMap()), new
RecordingConnectorContext());
+
+ DorisConnectorException listFailure =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> md.listPartitions(null, handle(), Optional.empty()));
+ Assertions.assertEquals("Metadata not found in metadata location for
table db1.t1",
+ listFailure.getMessage());
+ DorisConnectorException mvccFailure =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> md.getMvccPartitionView(null, handle()));
+ Assertions.assertEquals("Metadata not found in metadata location for
table db1.t1",
+ mvccFailure.getMessage());
+ }
}
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java
index 61f275f3593..02932e2d174 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergConnectorMetadataTest.java
@@ -35,6 +35,8 @@ import org.apache.iceberg.SortOrder;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.exceptions.NotFoundException;
import org.apache.iceberg.inmemory.InMemoryCatalog;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.view.ImmutableSQLViewRepresentation;
@@ -43,6 +45,7 @@ import org.apache.iceberg.view.ViewVersion;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.io.FileNotFoundException;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -89,6 +92,32 @@ public class IcebergConnectorMetadataTest {
Types.NestedField.optional(2, "name", Types.StringType.get()));
}
+ @Test
+ public void missingMetadataFileHasStableTableError() {
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ ops.loadTableFailure = new RuntimeException(
+ "Failed to open input stream", new
FileNotFoundException("missing metadata file"));
+
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> metadataWith(ops).getTableSchema(null, new
IcebergTableHandle("default", "missing_table")));
+ Assertions.assertTrue(ex.getMessage().contains(
+ "Metadata not found in metadata location for table
default.missing_table"), ex.getMessage());
+ }
+
+ @Test
+ public void missingMetadataFileForSystemTableHasStableTableError() {
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ ops.loadTableFailure = new RuntimeException(
+ "Failed to open input stream", new
FileNotFoundException("missing metadata file"));
+ IcebergTableHandle handle = IcebergTableHandle.forSystemTable(
+ "default", "missing_table", "snapshots", -1L, null, -1L);
+
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> metadataWith(ops).getTableSchema(null, handle));
+ Assertions.assertTrue(ex.getMessage().contains(
+ "Metadata not found in metadata location for table
default.missing_table"), ex.getMessage());
+ }
+
/**
* A view version whose summary records {@code engine-name} (when
non-null) and whose current version
* carries a SQL representation for {@code reprDialect} (when non-null) —
driving the sql/dialect extraction
@@ -579,6 +608,27 @@ public class IcebergConnectorMetadataTest {
Assertions.assertFalse(handleOpt.isPresent());
}
+ @Test
+ public void getTableHandleNormalizesDeepMetadataNotFoundFailure() {
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ ops.tableExistsFailure = new RuntimeException("catalog wrapper",
+ new RuntimeException("storage wrapper", new
NotFoundException("metadata is missing")));
+
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> metadataWith(ops).getTableHandle(null, "db1",
"missing_table"));
+ Assertions.assertEquals(
+ "Metadata not found in metadata location for table
db1.missing_table", ex.getMessage());
+ }
+
+ @Test
+ public void getTableHandlePreservesMissingTableAsEmpty() {
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ ops.tableExistsFailure = new RuntimeException("catalog wrapper",
+ new NoSuchTableException("table disappeared"));
+
+ Assertions.assertTrue(metadataWith(ops).getTableHandle(null, "db1",
"missing_table").isEmpty());
+ }
+
// ---------------------------------------------------------------------
// getTableSchema — load via seam, parse columns + table props
// ---------------------------------------------------------------------
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
index 9d0041bc602..911b6073a9b 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergScanPlanProviderTest.java
@@ -20,6 +20,7 @@ package org.apache.doris.connector.iceberg;
import org.apache.doris.connector.spi.ConnectorSession;
import org.apache.doris.connector.spi.ConnectorStatementScope;
import org.apache.doris.connector.spi.ConnectorType;
+import org.apache.doris.connector.spi.DorisConnectorException;
import org.apache.doris.connector.spi.handle.ConnectorColumnHandle;
import org.apache.doris.connector.spi.pushdown.ConnectorColumnRef;
import org.apache.doris.connector.spi.pushdown.ConnectorComparison;
@@ -73,6 +74,7 @@ import org.apache.iceberg.util.SerializationUtil;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.ZoneId;
@@ -209,6 +211,87 @@ public class IcebergScanPlanProviderTest {
Assertions.assertEquals("t1", ops.lastLoadTable);
}
+ @Test
+ public void planScanMissingMetadataFileHasStableTableError() {
+ RecordingIcebergCatalogOps ops = new RecordingIcebergCatalogOps();
+ ops.loadTableFailure = new RuntimeException(
+ "Failed to open input stream", new
FileNotFoundException("missing metadata file"));
+ IcebergScanPlanProvider provider = new IcebergScanPlanProvider(
+ IcebergCatalogProperties.of(Collections.emptyMap()), ops);
+
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> provider.planScan(null, ConnectorScanRequest.builder(
+ new IcebergTableHandle("default", "missing_table"),
Collections.emptyList()).build()));
+ Assertions.assertTrue(ex.getMessage().contains(
+ "Metadata not found in metadata location for table
default.missing_table"), ex.getMessage());
+ }
+
+ @Test
+ public void planScanMissingManifestListHasStableTableError() {
+ Table table = createTable("missing_manifest_list", SCHEMA,
PartitionSpec.unpartitioned());
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/missing_manifest_list/f1.parquet", 1024, null,
null)).commit();
+ table.io().deleteFile(table.currentSnapshot().manifestListLocation());
+ IcebergScanPlanProvider provider = providerOver(table);
+
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> provider.planScan(emptySession(),
ConnectorScanRequest.builder(
+ new IcebergTableHandle("db1",
"missing_manifest_list"), Collections.emptyList()).build()));
+ Assertions.assertTrue(ex.getMessage().contains(
+ "Metadata not found in metadata location for table
db1.missing_manifest_list"), ex.getMessage());
+ }
+
+ @Test
+ public void streamingMissingManifestListHasStableTableError() {
+ Table table = createTable("missing_stream_manifest", SCHEMA,
PartitionSpec.unpartitioned());
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/missing_stream_manifest/f1.parquet", 1024, null,
null)).commit();
+ table.io().deleteFile(table.currentSnapshot().manifestListLocation());
+ IcebergScanPlanProvider provider = providerOver(table);
+ IcebergTableHandle handle = new IcebergTableHandle("db1",
"missing_stream_manifest");
+
+ DorisConnectorException estimate =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> provider.streamingSplitEstimate(batchSession(1, true),
handle, Optional.empty(), false));
+ Assertions.assertTrue(estimate.getMessage().contains(
+ "Metadata not found in metadata location for table
db1.missing_stream_manifest"),
+ estimate.getMessage());
+
+ DorisConnectorException lazy =
Assertions.assertThrows(DorisConnectorException.class, () -> {
+ try (ConnectorSplitSource source = provider.streamSplits(
+ emptySession(), handle, Collections.emptyList(),
Optional.empty(), -1L)) {
+ source.hasNext();
+ }
+ });
+ Assertions.assertTrue(lazy.getMessage().contains(
+ "Metadata not found in metadata location for table
db1.missing_stream_manifest"),
+ lazy.getMessage());
+ }
+
+ @Test
+ public void scanPropertiesMissingDeleteManifestHasStableTableError() {
+ Schema schema = new Schema(
+ Arrays.asList(
+ Types.NestedField.required(1, "id",
Types.IntegerType.get()),
+ Types.NestedField.optional(2, "name",
Types.StringType.get())),
+ Collections.singleton(1));
+ Table table = createTable("missing_delete_manifest", schema,
PartitionSpec.unpartitioned(),
+ Collections.singletonMap(TableProperties.FORMAT_VERSION, "2"));
+ table.newAppend().appendFile(dataFile(table.spec(),
+ "s3://b/db/missing_delete_manifest/f1.parquet", 1024, null,
null)).commit();
+ table.newRowDelta().addDeletes(equalityDeleteFile(
+ "s3://b/db/missing_delete_manifest/eq.parquet",
FileFormat.PARQUET, 1)).commit();
+ String deleteManifest =
table.currentSnapshot().deleteManifests(table.io()).get(0).path().toString();
+ table.io().deleteFile(deleteManifest);
+ IcebergScanPlanProvider provider = providerOver(table);
+
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> provider.getScanNodeProperties(emptySession(),
+ new IcebergTableHandle("db1",
"missing_delete_manifest"),
+ Collections.singletonList(new
IcebergColumnHandle("name", 2)), Optional.empty()));
+ Assertions.assertTrue(ex.getMessage().contains(
+ "Metadata not found in metadata location for table
db1.missing_delete_manifest"), ex.getMessage());
+ }
+
@Test
public void planScanResolvesTableInsideAuthContext() {
// The remote loadTable must sit INSIDE context.executeAuthenticated
so the FE-injected Kerberos UGI
@@ -1358,6 +1441,23 @@ public class IcebergScanPlanProviderTest {
.build());
}
+ @Test
+ public void positionDeletesMissingManifestListHasStableTableError() {
+ DeleteFile deleteFile =
FileMetadata.deleteFileBuilder(PartitionSpec.unpartitioned())
+ .ofPositionDeletes()
+ .withPath("s3://b/db/t1/pos-delete.parquet")
+ .withFileSizeInBytes(100)
+ .withRecordCount(1)
+ .build();
+ Table table = tableWithPositionDelete(deleteFile);
+ table.io().deleteFile(table.currentSnapshot().manifestListLocation());
+
+ DorisConnectorException ex =
Assertions.assertThrows(DorisConnectorException.class,
+ () -> planPositionDeletes(table, Collections.emptyList()));
+ Assertions.assertTrue(ex.getMessage().contains(
+ "Metadata not found in metadata location for table db1.t1"),
ex.getMessage());
+ }
+
@Test
public void
planScanPositionDeletesEmitsNativeRangeMatchingTheBeRoutingContract() {
// WHY: this is THE contract. BE routes a range into
iceberg_position_delete_sys_table_reader iff
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/RecordingIcebergCatalogOps.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/RecordingIcebergCatalogOps.java
index 9f834eab01d..6a9526ea273 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/RecordingIcebergCatalogOps.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/RecordingIcebergCatalogOps.java
@@ -60,6 +60,8 @@ final class RecordingIcebergCatalogOps implements
IcebergCatalogOps {
boolean databaseExists;
/** Canned existence answer for {@link #tableExists(String, String)}. */
boolean tableExists;
+ /** Optional exact failure thrown by {@link #tableExists(String, String)}.
*/
+ RuntimeException tableExistsFailure;
/** Canned existence answer for {@link #viewExists(String, String)}. */
boolean viewExists;
/** Canned SDK view returned by {@link #loadView(String, String)}. */
@@ -74,6 +76,8 @@ final class RecordingIcebergCatalogOps implements
IcebergCatalogOps {
Table table;
/** When set, {@link #loadTable(String, String)} throws instead of
returning {@link #table}. */
boolean throwOnLoadTable;
+ /** Optional exact failure thrown by {@link #loadTable(String, String)}. */
+ RuntimeException loadTableFailure;
/** When set, {@link #loadTable(String, String)} throws {@link
NoSuchTableException} (concurrent-drop race). */
boolean throwNoSuchTableOnLoadTable;
/**
@@ -163,6 +167,9 @@ final class RecordingIcebergCatalogOps implements
IcebergCatalogOps {
log.add("tableExists:" + dbName + "." + tableName);
lastExistsDb = dbName;
lastExistsTable = tableName;
+ if (tableExistsFailure != null) {
+ throw tableExistsFailure;
+ }
return tableExists;
}
@@ -201,6 +208,9 @@ final class RecordingIcebergCatalogOps implements
IcebergCatalogOps {
if (throwNoSuchTableOnLoadTable) {
throw new NoSuchTableException("simulated missing table %s.%s",
dbName, tableName);
}
+ if (loadTableFailure != null) {
+ throw loadTableFailure;
+ }
if (throwOnLoadTable) {
throw new RuntimeException("simulated loadTable failure for " +
dbName + "." + tableName);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
index 7710b46224c..2328e34a00f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java
@@ -65,6 +65,7 @@ import
org.apache.doris.datasource.plugin.PluginDrivenMetadata;
import org.apache.doris.datasource.plugin.PluginDrivenSysExternalTable;
import org.apache.doris.datasource.split.FileSplit;
import org.apache.doris.datasource.split.PluginDrivenSplit;
+import org.apache.doris.nereids.exceptions.AnalysisException;
import
org.apache.doris.nereids.trees.plans.logical.LogicalFileScan.SelectedPartitions;
import org.apache.doris.planner.PlanNodeId;
import org.apache.doris.planner.ScanContext;
@@ -215,15 +216,17 @@ public class PluginDrivenScanNode extends
FileQueryScanNode {
Connector connector = catalog.getConnector();
ConnectorSession session = catalog.buildConnectorSession();
ConnectorMetadata metadata = PluginDrivenMetadata.get(session,
connector);
- String dbName = table.getDb() != null ? table.getDb().getRemoteName()
: "";
// Resolve through the table's sys-aware seam (NOT raw
metadata.getTableHandle): for a normal
// table this is identical to getTableHandle(session, dbName,
remoteName), but for a
// PluginDrivenSysExternalTable the override returns the connector's
SYSTEM handle (carrying
// sysTableName + forceJni), so the scan path threads force-JNI
correctly for binlog/audit_log.
ConnectorTableHandle handle =
table.resolveConnectorTableHandle(session, metadata)
- .orElseThrow(() -> new RuntimeException(
- "Table handle not found for plugin-driven table: " +
dbName + "."
- + table.getRemoteName()));
+ // Use analysis semantics and local names: mapped remote
identifiers are connector internals and
+ // a generic RuntimeException would be reported as
ERR_UNKNOWN_ERROR by EXPLAIN.
+ .orElseThrow(() -> new AnalysisException(
+ "Table '" + catalog.getName() + "."
+ + (table.getDb() == null ? "" :
table.getDb().getFullName()) + "." + table.getName()
+ + "' does not exist"));
return new PluginDrivenScanNode(id, desc, needCheckColumnPriv, sv,
scanContext, connector, session, handle);
}
@@ -1279,6 +1282,7 @@ public class PluginDrivenScanNode extends
FileQueryScanNode {
if (!(getTargetTable() instanceof PluginDrivenSysExternalTable)) {
return;
}
+ String connectorName = connectorDisplayName();
boolean timeTravelSupported = sysTableSupportsTimeTravel();
TableScanParams scanParams = getScanParams();
if (scanParams != null) {
@@ -1289,23 +1293,32 @@ public class PluginDrivenScanNode extends
FileQueryScanNode {
String sysTableName = sysTableName();
if (scanParams.incrementalRead()) {
if (!sysTableSupportsScanParam(p ->
p.supportsSystemTableIncrementalRead(sysTableName))) {
- throw new UserException("Plugin system table '" +
sysTableName
+ throw new UserException(connectorName + " system table '"
+ sysTableName
+ "' does not support INCR scan params.");
}
} else if (scanParams.isOptions()) {
if (!sysTableSupportsScanParam(p ->
p.supportsSystemTableOptions(sysTableName))) {
- throw new UserException("Plugin system table '" +
sysTableName
+ throw new UserException(connectorName + " system table '"
+ sysTableName
+ "' does not support OPTIONS scan params.");
}
} else if (!timeTravelSupported) {
- throw new UserException("Plugin system tables do not support
scan params.");
+ throw new UserException(connectorName + " system tables do not
support scan params.");
}
}
if (getQueryTableSnapshot() != null && !timeTravelSupported) {
- throw new UserException("Plugin system tables do not support time
travel.");
+ throw new UserException(connectorName + " system tables do not
support time travel.");
}
}
+ private String connectorDisplayName() throws UserException {
+ String engine = getTargetTable().getEngine();
+ // The engine is already the connector-owned display name; changing
its case corrupts identities such
+ // as iRODS and makes equivalent connector errors differ by execution
path.
+ return engine == null || engine.isEmpty()
+ ? "Plugin"
+ : engine;
+ }
+
/**
* Whether the connector honors THIS scan's selector on THIS system table
— the exact mirror of
* {@link #checkSysTableScanConstraints}' accept set, so a pin is resolved
for precisely the queries
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeSysHandleTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeSysHandleTest.java
index 6004ad6b8b8..a04072fd0bd 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeSysHandleTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeSysHandleTest.java
@@ -30,6 +30,7 @@ import org.apache.doris.datasource.SessionContext;
import org.apache.doris.datasource.plugin.PluginDrivenExternalCatalog;
import org.apache.doris.datasource.plugin.PluginDrivenExternalTable;
import org.apache.doris.datasource.plugin.PluginDrivenSysExternalTable;
+import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.planner.PlanNodeId;
import org.apache.doris.planner.ScanContext;
import org.apache.doris.qe.SessionVariable;
@@ -78,7 +79,7 @@ public class PluginDrivenScanNodeSysHandleTest {
ConnectorTableHandle sysHandle =
Mockito.mock(ConnectorTableHandle.class);
TestablePluginCatalog catalog = new TestablePluginCatalog("paimon",
metadata, session);
- ExternalDatabase<PluginDrivenExternalTable> db = mockDb("REMOTE_DB");
+ ExternalDatabase<PluginDrivenExternalTable> db = mockDb("local_db",
"REMOTE_DB");
// Base handle resolved from the SOURCE remote name (not the
"$"-suffixed sys remote name);
// the connector then maps base handle + "binlog" -> the sys handle.
NOTE: there is no stub
@@ -127,7 +128,7 @@ public class PluginDrivenScanNodeSysHandleTest {
ConnectorTableHandle baseHandle =
Mockito.mock(ConnectorTableHandle.class);
TestablePluginCatalog catalog = new TestablePluginCatalog("paimon",
metadata, session);
- ExternalDatabase<PluginDrivenExternalTable> db = mockDb("REMOTE_DB");
+ ExternalDatabase<PluginDrivenExternalTable> db = mockDb("local_db",
"REMOTE_DB");
Mockito.when(metadata.getTableHandle(session, "REMOTE_DB",
"REMOTE_TBL"))
.thenReturn(Optional.of(baseHandle));
@@ -148,6 +149,51 @@ public class PluginDrivenScanNodeSysHandleTest {
.getSysTableHandle(Mockito.any(), Mockito.any(),
Mockito.anyString());
}
+ @Test
+ public void createReportsMissingTableWhenHandleDisappears() {
+ ConnectorMetadata metadata = Mockito.mock(ConnectorMetadata.class);
+ ConnectorSession session = Mockito.mock(ConnectorSession.class);
+
Mockito.when(session.getStatementScope()).thenReturn(ConnectorStatementScope.NONE);
+ TestablePluginCatalog catalog = new TestablePluginCatalog("hms",
metadata, session);
+ ExternalDatabase<PluginDrivenExternalTable> db = mockDb("local_db",
"REMOTE_DB");
+ Mockito.when(metadata.getTableHandle(session, "REMOTE_DB",
"dropped_table"))
+ .thenReturn(Optional.empty());
+ PluginDrivenExternalTable table = bareTable(catalog, db,
"dropped_table");
+
+ AnalysisException ex = Assertions.assertThrows(AnalysisException.class,
+ () -> PluginDrivenScanNode.create(new PlanNodeId(0),
+ new TupleDescriptor(new TupleId(0)), false, new
SessionVariable(),
+ ScanContext.EMPTY, catalog, table));
+ Assertions.assertEquals("Table 'test-catalog.local_db.tbl' does not
exist", ex.getMessage());
+ }
+
+ @Test
+ public void createReportsMappedLocalNameWhenSysHandleDisappears() {
+ ConnectorMetadata metadata = Mockito.mock(ConnectorMetadata.class);
+ ConnectorSession session = Mockito.mock(ConnectorSession.class);
+
Mockito.when(session.getStatementScope()).thenReturn(ConnectorStatementScope.NONE);
+ ConnectorTableHandle baseHandle =
Mockito.mock(ConnectorTableHandle.class);
+ TestablePluginCatalog catalog = new TestablePluginCatalog("hms",
metadata, session);
+ ExternalDatabase<PluginDrivenExternalTable> db = mockDb("local_db",
"REMOTE_DB");
+ Mockito.when(metadata.getTableHandle(session, "REMOTE_DB",
"REMOTE_TBL"))
+ .thenReturn(Optional.of(baseHandle));
+ Mockito.when(metadata.getSysTableHandle(session, baseHandle, "files"))
+ .thenReturn(Optional.empty());
+ PluginDrivenExternalTable base = bareTable(catalog, db, "REMOTE_TBL");
+ PluginDrivenSysExternalTable sysTable = new
PluginDrivenSysExternalTable(base, "files") {
+ @Override
+ protected synchronized void makeSureInitialized() {
+ // no-op: skip Env-backed catalog/db init
+ }
+ };
+
+ AnalysisException ex = Assertions.assertThrows(AnalysisException.class,
+ () -> PluginDrivenScanNode.create(new PlanNodeId(0),
+ new TupleDescriptor(new TupleId(0)), false, new
SessionVariable(),
+ ScanContext.EMPTY, catalog, sysTable));
+ Assertions.assertEquals("Table 'test-catalog.local_db.tbl$files' does
not exist", ex.getMessage());
+ }
+
// ==================== helpers (mirror PluginDrivenSysTableTest)
====================
private static PluginDrivenExternalTable
bareTable(PluginDrivenExternalCatalog catalog,
@@ -161,8 +207,9 @@ public class PluginDrivenScanNodeSysHandleTest {
}
@SuppressWarnings("unchecked")
- private static ExternalDatabase<PluginDrivenExternalTable> mockDb(String
remoteName) {
+ private static ExternalDatabase<PluginDrivenExternalTable> mockDb(String
localName, String remoteName) {
ExternalDatabase<PluginDrivenExternalTable> db =
Mockito.mock(ExternalDatabase.class);
+ Mockito.when(db.getFullName()).thenReturn(localName);
Mockito.when(db.getRemoteName()).thenReturn(remoteName);
return db;
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeSysTableGuardTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeSysTableGuardTest.java
index 60247a27d22..0d034d4740a 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeSysTableGuardTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/scan/PluginDrivenScanNodeSysTableGuardTest.java
@@ -52,6 +52,13 @@ import org.mockito.Mockito;
*/
public class PluginDrivenScanNodeSysTableGuardTest {
+ private static PluginDrivenSysExternalTable sysTableWithEngine(String
engine) {
+ PluginDrivenSysExternalTable table =
Mockito.mock(PluginDrivenSysExternalTable.class);
+ Mockito.doReturn(engine).when(table).getEngine();
+ Mockito.doReturn("files").when(table).getSysTableName();
+ return table;
+ }
+
private static PluginDrivenScanNode guardOnlyNode() throws Exception {
PluginDrivenScanNode node = Mockito.mock(PluginDrivenScanNode.class,
Mockito.CALLS_REAL_METHODS);
// Default: no scan-params, no snapshot, and a connector whose sys
tables do NOT time-travel
@@ -93,6 +100,80 @@ public class PluginDrivenScanNodeSysTableGuardTest {
"time-travel rejection must carry the expected message, got: "
+ ex.getMessage());
}
+ @Test
+ public void sysTableTimeTravelErrorNamesConnector() throws Exception {
+ PluginDrivenScanNode node = guardOnlyNode();
+ PluginDrivenSysExternalTable sysTable =
Mockito.mock(PluginDrivenSysExternalTable.class);
+ Mockito.doReturn("paimon").when(sysTable).getEngine();
+ Mockito.doReturn(sysTable).when(node).getTargetTable();
+
Mockito.doReturn(Mockito.mock(TableSnapshot.class)).when(node).getQueryTableSnapshot();
+
+ UserException ex = Assertions.assertThrows(UserException.class,
+ node::checkSysTableScanConstraints);
+ Assertions.assertTrue(ex.getMessage().contains("paimon system tables
do not support time travel."),
+ ex.getMessage());
+ }
+
+ @Test
+ public void sysTableScanParamsErrorNamesConnector() throws Exception {
+ PluginDrivenScanNode node = guardOnlyNode();
+ PluginDrivenSysExternalTable sysTable =
Mockito.mock(PluginDrivenSysExternalTable.class);
+ Mockito.doReturn("paimon").when(sysTable).getEngine();
+ Mockito.doReturn(sysTable).when(node).getTargetTable();
+
Mockito.doReturn(Mockito.mock(TableScanParams.class)).when(node).getScanParams();
+
+ UserException ex = Assertions.assertThrows(UserException.class,
+ node::checkSysTableScanConstraints);
+ Assertions.assertTrue(ex.getMessage().contains("paimon system tables
do not support scan params."),
+ ex.getMessage());
+ }
+
+ @Test
+ public void mixedCaseConnectorNameIsPreservedForIncrementalReadError()
throws Exception {
+ PluginDrivenScanNode node = guardOnlyNode();
+
Mockito.doReturn(sysTableWithEngine("iRODS")).when(node).getTargetTable();
+ TableScanParams incr = Mockito.mock(TableScanParams.class);
+ Mockito.doReturn(true).when(incr).incrementalRead();
+ Mockito.doReturn(incr).when(node).getScanParams();
+
+ UserException ex = Assertions.assertThrows(UserException.class,
node::checkSysTableScanConstraints);
+ Assertions.assertEquals(
+ "iRODS system table 'files' does not support INCR scan
params.", ex.getDetailMessage());
+ }
+
+ @Test
+ public void mixedCaseConnectorNameIsPreservedForOptionsError() throws
Exception {
+ PluginDrivenScanNode node = guardOnlyNode();
+
Mockito.doReturn(sysTableWithEngine("iRODS")).when(node).getTargetTable();
+ TableScanParams options = Mockito.mock(TableScanParams.class);
+ Mockito.doReturn(true).when(options).isOptions();
+ Mockito.doReturn(options).when(node).getScanParams();
+
+ UserException ex = Assertions.assertThrows(UserException.class,
node::checkSysTableScanConstraints);
+ Assertions.assertEquals(
+ "iRODS system table 'files' does not support OPTIONS scan
params.", ex.getDetailMessage());
+ }
+
+ @Test
+ public void mixedCaseConnectorNameIsPreservedForGenericScanParamsError()
throws Exception {
+ PluginDrivenScanNode node = guardOnlyNode();
+
Mockito.doReturn(sysTableWithEngine("iRODS")).when(node).getTargetTable();
+
Mockito.doReturn(Mockito.mock(TableScanParams.class)).when(node).getScanParams();
+
+ UserException ex = Assertions.assertThrows(UserException.class,
node::checkSysTableScanConstraints);
+ Assertions.assertEquals("iRODS system tables do not support scan
params.", ex.getDetailMessage());
+ }
+
+ @Test
+ public void mixedCaseConnectorNameIsPreservedForTimeTravelError() throws
Exception {
+ PluginDrivenScanNode node = guardOnlyNode();
+
Mockito.doReturn(sysTableWithEngine("iRODS")).when(node).getTargetTable();
+
Mockito.doReturn(Mockito.mock(TableSnapshot.class)).when(node).getQueryTableSnapshot();
+
+ UserException ex = Assertions.assertThrows(UserException.class,
node::checkSysTableScanConstraints);
+ Assertions.assertEquals("iRODS system tables do not support time
travel.", ex.getDetailMessage());
+ }
+
@Test
public void sysTableWithoutScanParamsOrSnapshotDoesNotThrow() throws
Exception {
PluginDrivenScanNode node = guardOnlyNode();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
