You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "ahmedabu98 (via GitHub)" <gi...@apache.org> on 2025/06/10 20:24:16 UTC
[PR] [IcebergIO] Create namespaces if needed [beam]
ahmedabu98 opened a new pull request, #35228:
URL: https://github.com/apache/beam/pull/35228
Dynamically creates a namespace if it does not exist yet.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [IcebergIO] Create namespaces if needed [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on code in PR #35228:
URL: https://github.com/apache/beam/pull/35228#discussion_r2140260220
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java:
##########
@@ -851,6 +857,113 @@ public void testStreamToPartitionedDynamicDestinations() throws IOException {
writeToDynamicDestinations(null, true, true);
}
+ @Test
+ public void testWriteToDynamicNamespaces() throws IOException {
+ // run this test only on catalogs that support namespace management
+ assumeTrue(catalog instanceof SupportsNamespaces);
+
+ long salt = System.currentTimeMillis();
+ String tableIdentifierTemplate =
+ String.format("namespace_{modulo_5}_%s.table_{bool_field}", salt);
+ Map<String, Object> writeConfig = new HashMap<>(managedIcebergConfig(tableIdentifierTemplate));
+ // override with table template
+ writeConfig.put("table", tableIdentifierTemplate);
+
+ Namespace namespace0 = Namespace.of("namespace_0_" + salt);
+ Namespace namespace1 = Namespace.of("namespace_1_" + salt);
+ Namespace namespace2 = Namespace.of("namespace_2_" + salt);
+ Namespace namespace3 = Namespace.of("namespace_3_" + salt);
+ Namespace namespace4 = Namespace.of("namespace_4_" + salt);
+
+ TableIdentifier tableId0true = TableIdentifier.of(namespace0, "table_true");
+ TableIdentifier tableId0false = TableIdentifier.of(namespace0, "table_false");
+ TableIdentifier tableId1true = TableIdentifier.of(namespace1, "table_true");
+ TableIdentifier tableId1false = TableIdentifier.of(namespace1, "table_false");
+ TableIdentifier tableId2true = TableIdentifier.of(namespace2, "table_true");
+ TableIdentifier tableId2false = TableIdentifier.of(namespace2, "table_false");
+ TableIdentifier tableId3true = TableIdentifier.of(namespace3, "table_true");
+ TableIdentifier tableId3false = TableIdentifier.of(namespace3, "table_false");
+ TableIdentifier tableId4true = TableIdentifier.of(namespace4, "table_true");
+ TableIdentifier tableId4false = TableIdentifier.of(namespace4, "table_false");
+
+ List<Namespace> namespaces =
+ Arrays.asList(namespace0, namespace1, namespace2, namespace3, namespace4);
+ SupportsNamespaces sN = (SupportsNamespaces) catalog;
+ // assert namespace don't exist beforehand
+ namespaces.forEach(n -> assertFalse(sN.namespaceExists(n)));
+
+ pipeline
+ .apply(Create.of(inputRows))
+ .setRowSchema(BEAM_SCHEMA)
+ .apply(Managed.write(ICEBERG).withConfig(writeConfig));
+ pipeline.run().waitUntilFinish();
+
+ // assert namespace were created
+ namespaces.forEach(n -> assertTrue(sN.namespaceExists(n)));
+
+ Table table0true = catalog.loadTable(tableId0true);
+ Table table0false = catalog.loadTable(tableId0false);
+ Table table1true = catalog.loadTable(tableId1true);
+ Table table1false = catalog.loadTable(tableId1false);
+ Table table2true = catalog.loadTable(tableId2true);
+ Table table2false = catalog.loadTable(tableId2false);
+ Table table3true = catalog.loadTable(tableId3true);
+ Table table3false = catalog.loadTable(tableId3false);
+ Table table4true = catalog.loadTable(tableId4true);
+ Table table4false = catalog.loadTable(tableId4false);
+
+ for (Table t :
+ Arrays.asList(
+ table0true,
+ table0false,
+ table1true,
+ table1false,
+ table2true,
+ table2false,
+ table3true,
+ table3false,
+ table4true,
+ table4false)) {
+ assertTrue(t.schema().sameSchema(ICEBERG_SCHEMA));
+ }
+
+ // Read back and check records are correct
+ Map<KV<Long, Boolean>, List<Record>> results =
+ ImmutableMap.<KV<Long, Boolean>, List<Record>>builder()
+ .put(KV.of(0L, true), readRecords(table0true))
+ .put(KV.of(0L, false), readRecords(table0false))
+ .put(KV.of(1L, true), readRecords(table1true))
+ .put(KV.of(1L, false), readRecords(table1false))
+ .put(KV.of(2L, true), readRecords(table2true))
+ .put(KV.of(2L, false), readRecords(table2false))
+ .put(KV.of(3L, true), readRecords(table3true))
+ .put(KV.of(3L, false), readRecords(table3false))
+ .put(KV.of(4L, true), readRecords(table4true))
+ .put(KV.of(4L, false), readRecords(table4false))
+ .build();
+
+ for (Map.Entry<KV<Long, Boolean>, List<Record>> entry : results.entrySet()) {
+ long modulo = entry.getKey().getKey();
+ boolean bool = entry.getKey().getValue();
+ List<Record> records = entry.getValue();
+ Stream<Record> expectedRecords =
+ inputRows.stream()
+ .filter(
+ rec ->
+ checkStateNotNull(rec.getInt64("modulo_5")) == modulo
+ && checkStateNotNull(rec.getBoolean("bool_field")) == bool)
+ .map(RECORD_FUNC::apply);
+
+ assertThat(records, containsInAnyOrder(expectedRecords.toArray()));
+ }
+
+ try {
+ namespaces.forEach(sN::dropNamespace);
+ } catch (Exception e) {
+ LOG.error("Test passed but threw an error when cleaning up namespaces.", e);
Review Comment:
Done, will merge when tests go green
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [IcebergIO] Create namespaces if needed [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on PR #35228:
URL: https://github.com/apache/beam/pull/35228#issuecomment-2962184303
Can you file issues or comment on the flakes? Or even add `@Ignore` flags to tests that are problematic? I'd rather get in the habit of merging on green and keeping clear signal.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [IcebergIO] Create namespaces if needed [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #35228:
URL: https://github.com/apache/beam/pull/35228#issuecomment-2960651023
Assigning reviewers:
R: @m-trieu for label java.
Note: If you would like to opt out of this review, comment `assign to next reviewer`.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [IcebergIO] Create namespaces if needed [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on code in PR #35228:
URL: https://github.com/apache/beam/pull/35228#discussion_r2139839244
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java:
##########
@@ -851,6 +857,113 @@ public void testStreamToPartitionedDynamicDestinations() throws IOException {
writeToDynamicDestinations(null, true, true);
}
+ @Test
+ public void testWriteToDynamicNamespaces() throws IOException {
+ // run this test only on catalogs that support namespace management
+ assumeTrue(catalog instanceof SupportsNamespaces);
+
+ long salt = System.currentTimeMillis();
+ String tableIdentifierTemplate =
+ String.format("namespace_{modulo_5}_%s.table_{bool_field}", salt);
+ Map<String, Object> writeConfig = new HashMap<>(managedIcebergConfig(tableIdentifierTemplate));
+ // override with table template
+ writeConfig.put("table", tableIdentifierTemplate);
+
+ Namespace namespace0 = Namespace.of("namespace_0_" + salt);
+ Namespace namespace1 = Namespace.of("namespace_1_" + salt);
+ Namespace namespace2 = Namespace.of("namespace_2_" + salt);
+ Namespace namespace3 = Namespace.of("namespace_3_" + salt);
+ Namespace namespace4 = Namespace.of("namespace_4_" + salt);
+
+ TableIdentifier tableId0true = TableIdentifier.of(namespace0, "table_true");
+ TableIdentifier tableId0false = TableIdentifier.of(namespace0, "table_false");
+ TableIdentifier tableId1true = TableIdentifier.of(namespace1, "table_true");
+ TableIdentifier tableId1false = TableIdentifier.of(namespace1, "table_false");
+ TableIdentifier tableId2true = TableIdentifier.of(namespace2, "table_true");
+ TableIdentifier tableId2false = TableIdentifier.of(namespace2, "table_false");
+ TableIdentifier tableId3true = TableIdentifier.of(namespace3, "table_true");
+ TableIdentifier tableId3false = TableIdentifier.of(namespace3, "table_false");
+ TableIdentifier tableId4true = TableIdentifier.of(namespace4, "table_true");
+ TableIdentifier tableId4false = TableIdentifier.of(namespace4, "table_false");
+
+ List<Namespace> namespaces =
+ Arrays.asList(namespace0, namespace1, namespace2, namespace3, namespace4);
+ SupportsNamespaces sN = (SupportsNamespaces) catalog;
+ // assert namespace don't exist beforehand
+ namespaces.forEach(n -> assertFalse(sN.namespaceExists(n)));
+
+ pipeline
+ .apply(Create.of(inputRows))
+ .setRowSchema(BEAM_SCHEMA)
+ .apply(Managed.write(ICEBERG).withConfig(writeConfig));
+ pipeline.run().waitUntilFinish();
+
+ // assert namespace were created
+ namespaces.forEach(n -> assertTrue(sN.namespaceExists(n)));
+
+ Table table0true = catalog.loadTable(tableId0true);
+ Table table0false = catalog.loadTable(tableId0false);
+ Table table1true = catalog.loadTable(tableId1true);
+ Table table1false = catalog.loadTable(tableId1false);
+ Table table2true = catalog.loadTable(tableId2true);
+ Table table2false = catalog.loadTable(tableId2false);
+ Table table3true = catalog.loadTable(tableId3true);
+ Table table3false = catalog.loadTable(tableId3false);
+ Table table4true = catalog.loadTable(tableId4true);
+ Table table4false = catalog.loadTable(tableId4false);
+
+ for (Table t :
+ Arrays.asList(
+ table0true,
+ table0false,
+ table1true,
+ table1false,
+ table2true,
+ table2false,
+ table3true,
+ table3false,
+ table4true,
+ table4false)) {
+ assertTrue(t.schema().sameSchema(ICEBERG_SCHEMA));
+ }
+
+ // Read back and check records are correct
+ Map<KV<Long, Boolean>, List<Record>> results =
+ ImmutableMap.<KV<Long, Boolean>, List<Record>>builder()
+ .put(KV.of(0L, true), readRecords(table0true))
+ .put(KV.of(0L, false), readRecords(table0false))
+ .put(KV.of(1L, true), readRecords(table1true))
+ .put(KV.of(1L, false), readRecords(table1false))
+ .put(KV.of(2L, true), readRecords(table2true))
+ .put(KV.of(2L, false), readRecords(table2false))
+ .put(KV.of(3L, true), readRecords(table3true))
+ .put(KV.of(3L, false), readRecords(table3false))
+ .put(KV.of(4L, true), readRecords(table4true))
+ .put(KV.of(4L, false), readRecords(table4false))
+ .build();
+
+ for (Map.Entry<KV<Long, Boolean>, List<Record>> entry : results.entrySet()) {
+ long modulo = entry.getKey().getKey();
+ boolean bool = entry.getKey().getValue();
+ List<Record> records = entry.getValue();
+ Stream<Record> expectedRecords =
+ inputRows.stream()
+ .filter(
+ rec ->
+ checkStateNotNull(rec.getInt64("modulo_5")) == modulo
+ && checkStateNotNull(rec.getBoolean("bool_field")) == bool)
+ .map(RECORD_FUNC::apply);
+
+ assertThat(records, containsInAnyOrder(expectedRecords.toArray()));
+ }
+
+ try {
+ namespaces.forEach(sN::dropNamespace);
+ } catch (Exception e) {
+ LOG.error("Test passed but threw an error when cleaning up namespaces.", e);
Review Comment:
It fails pretty consistently with the BigQueryMetastore catalog. Agreed we can prob do more to fully cleanup, I can add some tweaking to fall back on BQ client to delete datasets.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [IcebergIO] Create namespaces if needed [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on code in PR #35228:
URL: https://github.com/apache/beam/pull/35228#discussion_r2139816015
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java:
##########
@@ -851,6 +857,113 @@ public void testStreamToPartitionedDynamicDestinations() throws IOException {
writeToDynamicDestinations(null, true, true);
}
+ @Test
+ public void testWriteToDynamicNamespaces() throws IOException {
+ // run this test only on catalogs that support namespace management
+ assumeTrue(catalog instanceof SupportsNamespaces);
+
+ long salt = System.currentTimeMillis();
+ String tableIdentifierTemplate =
+ String.format("namespace_{modulo_5}_%s.table_{bool_field}", salt);
+ Map<String, Object> writeConfig = new HashMap<>(managedIcebergConfig(tableIdentifierTemplate));
+ // override with table template
+ writeConfig.put("table", tableIdentifierTemplate);
+
+ Namespace namespace0 = Namespace.of("namespace_0_" + salt);
+ Namespace namespace1 = Namespace.of("namespace_1_" + salt);
+ Namespace namespace2 = Namespace.of("namespace_2_" + salt);
+ Namespace namespace3 = Namespace.of("namespace_3_" + salt);
+ Namespace namespace4 = Namespace.of("namespace_4_" + salt);
+
+ TableIdentifier tableId0true = TableIdentifier.of(namespace0, "table_true");
+ TableIdentifier tableId0false = TableIdentifier.of(namespace0, "table_false");
+ TableIdentifier tableId1true = TableIdentifier.of(namespace1, "table_true");
+ TableIdentifier tableId1false = TableIdentifier.of(namespace1, "table_false");
+ TableIdentifier tableId2true = TableIdentifier.of(namespace2, "table_true");
+ TableIdentifier tableId2false = TableIdentifier.of(namespace2, "table_false");
+ TableIdentifier tableId3true = TableIdentifier.of(namespace3, "table_true");
+ TableIdentifier tableId3false = TableIdentifier.of(namespace3, "table_false");
+ TableIdentifier tableId4true = TableIdentifier.of(namespace4, "table_true");
+ TableIdentifier tableId4false = TableIdentifier.of(namespace4, "table_false");
+
+ List<Namespace> namespaces =
+ Arrays.asList(namespace0, namespace1, namespace2, namespace3, namespace4);
+ SupportsNamespaces sN = (SupportsNamespaces) catalog;
+ // assert namespace don't exist beforehand
+ namespaces.forEach(n -> assertFalse(sN.namespaceExists(n)));
+
+ pipeline
+ .apply(Create.of(inputRows))
+ .setRowSchema(BEAM_SCHEMA)
+ .apply(Managed.write(ICEBERG).withConfig(writeConfig));
+ pipeline.run().waitUntilFinish();
+
+ // assert namespace were created
+ namespaces.forEach(n -> assertTrue(sN.namespaceExists(n)));
+
+ Table table0true = catalog.loadTable(tableId0true);
+ Table table0false = catalog.loadTable(tableId0false);
+ Table table1true = catalog.loadTable(tableId1true);
+ Table table1false = catalog.loadTable(tableId1false);
+ Table table2true = catalog.loadTable(tableId2true);
+ Table table2false = catalog.loadTable(tableId2false);
+ Table table3true = catalog.loadTable(tableId3true);
+ Table table3false = catalog.loadTable(tableId3false);
+ Table table4true = catalog.loadTable(tableId4true);
+ Table table4false = catalog.loadTable(tableId4false);
+
+ for (Table t :
+ Arrays.asList(
+ table0true,
+ table0false,
+ table1true,
+ table1false,
+ table2true,
+ table2false,
+ table3true,
+ table3false,
+ table4true,
+ table4false)) {
+ assertTrue(t.schema().sameSchema(ICEBERG_SCHEMA));
+ }
+
+ // Read back and check records are correct
+ Map<KV<Long, Boolean>, List<Record>> results =
+ ImmutableMap.<KV<Long, Boolean>, List<Record>>builder()
+ .put(KV.of(0L, true), readRecords(table0true))
+ .put(KV.of(0L, false), readRecords(table0false))
+ .put(KV.of(1L, true), readRecords(table1true))
+ .put(KV.of(1L, false), readRecords(table1false))
+ .put(KV.of(2L, true), readRecords(table2true))
+ .put(KV.of(2L, false), readRecords(table2false))
+ .put(KV.of(3L, true), readRecords(table3true))
+ .put(KV.of(3L, false), readRecords(table3false))
+ .put(KV.of(4L, true), readRecords(table4true))
+ .put(KV.of(4L, false), readRecords(table4false))
+ .build();
+
+ for (Map.Entry<KV<Long, Boolean>, List<Record>> entry : results.entrySet()) {
+ long modulo = entry.getKey().getKey();
+ boolean bool = entry.getKey().getValue();
+ List<Record> records = entry.getValue();
+ Stream<Record> expectedRecords =
+ inputRows.stream()
+ .filter(
+ rec ->
+ checkStateNotNull(rec.getInt64("modulo_5")) == modulo
+ && checkStateNotNull(rec.getBoolean("bool_field")) == bool)
+ .map(RECORD_FUNC::apply);
+
+ assertThat(records, containsInAnyOrder(expectedRecords.toArray()));
+ }
+
+ try {
+ namespaces.forEach(sN::dropNamespace);
+ } catch (Exception e) {
+ LOG.error("Test passed but threw an error when cleaning up namespaces.", e);
Review Comment:
Does this happen often? I'm not convinced (without some rationale) that we should put logic in to ignore this.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [IcebergIO] Create namespaces if needed [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 merged PR #35228:
URL: https://github.com/apache/beam/pull/35228
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [IcebergIO] Create namespaces if needed [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on code in PR #35228:
URL: https://github.com/apache/beam/pull/35228#discussion_r2139839244
##########
sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/catalog/IcebergCatalogBaseIT.java:
##########
@@ -851,6 +857,113 @@ public void testStreamToPartitionedDynamicDestinations() throws IOException {
writeToDynamicDestinations(null, true, true);
}
+ @Test
+ public void testWriteToDynamicNamespaces() throws IOException {
+ // run this test only on catalogs that support namespace management
+ assumeTrue(catalog instanceof SupportsNamespaces);
+
+ long salt = System.currentTimeMillis();
+ String tableIdentifierTemplate =
+ String.format("namespace_{modulo_5}_%s.table_{bool_field}", salt);
+ Map<String, Object> writeConfig = new HashMap<>(managedIcebergConfig(tableIdentifierTemplate));
+ // override with table template
+ writeConfig.put("table", tableIdentifierTemplate);
+
+ Namespace namespace0 = Namespace.of("namespace_0_" + salt);
+ Namespace namespace1 = Namespace.of("namespace_1_" + salt);
+ Namespace namespace2 = Namespace.of("namespace_2_" + salt);
+ Namespace namespace3 = Namespace.of("namespace_3_" + salt);
+ Namespace namespace4 = Namespace.of("namespace_4_" + salt);
+
+ TableIdentifier tableId0true = TableIdentifier.of(namespace0, "table_true");
+ TableIdentifier tableId0false = TableIdentifier.of(namespace0, "table_false");
+ TableIdentifier tableId1true = TableIdentifier.of(namespace1, "table_true");
+ TableIdentifier tableId1false = TableIdentifier.of(namespace1, "table_false");
+ TableIdentifier tableId2true = TableIdentifier.of(namespace2, "table_true");
+ TableIdentifier tableId2false = TableIdentifier.of(namespace2, "table_false");
+ TableIdentifier tableId3true = TableIdentifier.of(namespace3, "table_true");
+ TableIdentifier tableId3false = TableIdentifier.of(namespace3, "table_false");
+ TableIdentifier tableId4true = TableIdentifier.of(namespace4, "table_true");
+ TableIdentifier tableId4false = TableIdentifier.of(namespace4, "table_false");
+
+ List<Namespace> namespaces =
+ Arrays.asList(namespace0, namespace1, namespace2, namespace3, namespace4);
+ SupportsNamespaces sN = (SupportsNamespaces) catalog;
+ // assert namespace don't exist beforehand
+ namespaces.forEach(n -> assertFalse(sN.namespaceExists(n)));
+
+ pipeline
+ .apply(Create.of(inputRows))
+ .setRowSchema(BEAM_SCHEMA)
+ .apply(Managed.write(ICEBERG).withConfig(writeConfig));
+ pipeline.run().waitUntilFinish();
+
+ // assert namespace were created
+ namespaces.forEach(n -> assertTrue(sN.namespaceExists(n)));
+
+ Table table0true = catalog.loadTable(tableId0true);
+ Table table0false = catalog.loadTable(tableId0false);
+ Table table1true = catalog.loadTable(tableId1true);
+ Table table1false = catalog.loadTable(tableId1false);
+ Table table2true = catalog.loadTable(tableId2true);
+ Table table2false = catalog.loadTable(tableId2false);
+ Table table3true = catalog.loadTable(tableId3true);
+ Table table3false = catalog.loadTable(tableId3false);
+ Table table4true = catalog.loadTable(tableId4true);
+ Table table4false = catalog.loadTable(tableId4false);
+
+ for (Table t :
+ Arrays.asList(
+ table0true,
+ table0false,
+ table1true,
+ table1false,
+ table2true,
+ table2false,
+ table3true,
+ table3false,
+ table4true,
+ table4false)) {
+ assertTrue(t.schema().sameSchema(ICEBERG_SCHEMA));
+ }
+
+ // Read back and check records are correct
+ Map<KV<Long, Boolean>, List<Record>> results =
+ ImmutableMap.<KV<Long, Boolean>, List<Record>>builder()
+ .put(KV.of(0L, true), readRecords(table0true))
+ .put(KV.of(0L, false), readRecords(table0false))
+ .put(KV.of(1L, true), readRecords(table1true))
+ .put(KV.of(1L, false), readRecords(table1false))
+ .put(KV.of(2L, true), readRecords(table2true))
+ .put(KV.of(2L, false), readRecords(table2false))
+ .put(KV.of(3L, true), readRecords(table3true))
+ .put(KV.of(3L, false), readRecords(table3false))
+ .put(KV.of(4L, true), readRecords(table4true))
+ .put(KV.of(4L, false), readRecords(table4false))
+ .build();
+
+ for (Map.Entry<KV<Long, Boolean>, List<Record>> entry : results.entrySet()) {
+ long modulo = entry.getKey().getKey();
+ boolean bool = entry.getKey().getValue();
+ List<Record> records = entry.getValue();
+ Stream<Record> expectedRecords =
+ inputRows.stream()
+ .filter(
+ rec ->
+ checkStateNotNull(rec.getInt64("modulo_5")) == modulo
+ && checkStateNotNull(rec.getBoolean("bool_field")) == bool)
+ .map(RECORD_FUNC::apply);
+
+ assertThat(records, containsInAnyOrder(expectedRecords.toArray()));
+ }
+
+ try {
+ namespaces.forEach(sN::dropNamespace);
+ } catch (Exception e) {
+ LOG.error("Test passed but threw an error when cleaning up namespaces.", e);
Review Comment:
~It fails pretty consistently with the BigQueryMetastore catalog. Agreed we can prob do more to fully cleanup, I can add some tweaking to fall back on BQ client to delete datasets~.
Turns out I just wasn't clearing up tables before dropping the namespace
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [IcebergIO] Create namespaces if needed [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on PR #35228:
URL: https://github.com/apache/beam/pull/35228#issuecomment-2962330454
(#31590)
- Java Unit Tests (self-hosted, ubuntu-20.04, main) --> `UnboundedScheduledExecutorServiceTest.testThreadsAreAddedOnlyAsNeededWithContention`
(#35233)
- Java Precommit --> `PrismRunnerTest.windowing`
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [IcebergIO] Create namespaces if needed [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #35228:
URL: https://github.com/apache/beam/pull/35228#issuecomment-2960640725
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment `assign set of reviewers`
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [IcebergIO] Create namespaces if needed [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on PR #35228:
URL: https://github.com/apache/beam/pull/35228#issuecomment-2960649623
Test failures are not relevant
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [IcebergIO] Create namespaces if needed [beam]
Posted by "ahmedabu98 (via GitHub)" <gi...@apache.org>.
ahmedabu98 commented on PR #35228:
URL: https://github.com/apache/beam/pull/35228#issuecomment-2960649108
assign set of reviewers
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org