You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by vt...@apache.org on 2025/07/22 05:45:30 UTC
(beam) branch fix-iceberg-it created (now e77e8c3ff4a)
This is an automated email from the ASF dual-hosted git repository.
vterentev pushed a change to branch fix-iceberg-it
in repository https://gitbox.apache.org/repos/asf/beam.git
at e77e8c3ff4a Revert #35567
This branch includes the following new commits:
new e77e8c3ff4a Revert #35567
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(beam) 01/01: Revert #35567
Posted by vt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vterentev pushed a commit to branch fix-iceberg-it
in repository https://gitbox.apache.org/repos/asf/beam.git
commit e77e8c3ff4a55b4ba2d7016184e2f24c44b34001
Author: Vitaly Terentyev <vi...@akvelon.com>
AuthorDate: Tue Jul 22 09:45:15 2025 +0400
Revert #35567
---
.../sdk/io/gcp/bigquery/BigQueryServicesImpl.java | 105 +++------------------
.../io/gcp/bigquery/BigQueryServicesImplTest.java | 80 +---------------
2 files changed, 13 insertions(+), 172 deletions(-)
diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java
index f4303886c7a..390ffa1aa99 100644
--- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java
+++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java
@@ -109,14 +109,12 @@ import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.Callable;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
-import java.util.stream.Collectors;
import org.apache.beam.fn.harness.logging.QuotaEvent;
import org.apache.beam.fn.harness.logging.QuotaEvent.QuotaEventCloseable;
import org.apache.beam.runners.core.metrics.MonitoringInfoConstants;
@@ -174,12 +172,7 @@ public class BigQueryServicesImpl implements BigQueryServices {
// The approximate maximum payload of rows for an insertAll request.
// We set it to 9MB, which leaves room for request overhead.
- private static final Integer MAX_BQ_ROW_PAYLOAD_MB = 9;
-
- private static final Integer MAX_BQ_ROW_PAYLOAD_BYTES = MAX_BQ_ROW_PAYLOAD_MB * 1024 * 1024;
-
- private static final String MAX_BQ_ROW_PAYLOAD_DESC =
- String.format("%sMB", MAX_BQ_ROW_PAYLOAD_MB);
+ private static final Integer MAX_BQ_ROW_PAYLOAD = 9 * 1024 * 1024;
// The initial backoff for polling the status of a BigQuery job.
private static final Duration INITIAL_JOB_STATUS_POLL_BACKOFF = Duration.standardSeconds(1);
@@ -603,7 +596,6 @@ public class BigQueryServicesImpl implements BigQueryServices {
private final PipelineOptions options;
private final long maxRowsPerBatch;
private final long maxRowBatchSize;
- private final Map<String, TableSchema> tableSchemaCache = new ConcurrentHashMap<>();
// aggregate the total time spent in exponential backoff
private final Counter throttlingMsecs =
Metrics.counter(DatasetServiceImpl.class, Metrics.THROTTLE_TIME_COUNTER_NAME);
@@ -1157,53 +1149,22 @@ public class BigQueryServicesImpl implements BigQueryServices {
// If this row's encoding by itself is larger than the maximum row payload, then it's
// impossible to insert into BigQuery, and so we send it out through the dead-letter
// queue.
- if (nextRowSize >= MAX_BQ_ROW_PAYLOAD_BYTES) {
+ if (nextRowSize >= MAX_BQ_ROW_PAYLOAD) {
InsertErrors error =
new InsertErrors()
.setErrors(ImmutableList.of(new ErrorProto().setReason("row-too-large")));
// We verify whether the retryPolicy parameter expects us to retry. If it does, then
// it will return true. Otherwise it will return false.
- if (retryPolicy.shouldRetry(new InsertRetryPolicy.Context(error))) {
- // Obtain table schema
- TableSchema tableSchema = null;
- try {
- String tableSpec = BigQueryHelpers.toTableSpec(ref);
- if (tableSchemaCache.containsKey(tableSpec)) {
- tableSchema = tableSchemaCache.get(tableSpec);
- } else {
- Table table = getTable(ref);
- if (table != null) {
- tableSchema =
- TableRowToStorageApiProto.schemaToProtoTableSchema(table.getSchema());
- tableSchemaCache.put(tableSpec, tableSchema);
- }
- }
- } catch (Exception e) {
- LOG.warn("Failed to get table schema", e);
- }
-
- // Validate row schema
- String rowDetails = "";
- if (tableSchema != null) {
- rowDetails = validateRowSchema(row, tableSchema);
- }
-
- // Basic log to return
- String bqLimitLog =
+ Boolean isRetry = retryPolicy.shouldRetry(new InsertRetryPolicy.Context(error));
+ if (isRetry) {
+ throw new RuntimeException(
String.format(
- "We have observed a row of size %s bytes exceeding the "
- + "BigQueryIO limit of %s.",
- nextRowSize, MAX_BQ_ROW_PAYLOAD_DESC);
-
- // Add on row schema diff details if present
- if (!rowDetails.isEmpty()) {
- bqLimitLog +=
- String.format(
- " This is probably due to a schema "
- + "mismatch. Problematic row had extra schema fields: %s.",
- rowDetails);
- }
- throw new RuntimeException(bqLimitLog);
+ "We have observed a row that is %s bytes in size and exceeded BigQueryIO"
+ + " limit of 9MB. While BigQuery supports request sizes up to 10MB,"
+ + " BigQueryIO sets the limit at 9MB to leave room for request"
+ + " overhead. You may change your retry strategy to unblock this"
+ + " pipeline, and the row will be output as a failed insert.",
+ nextRowSize));
} else {
numFailedRows += 1;
errorContainer.add(failedInserts, error, ref, rowsToPublish.get(rowIndex));
@@ -1216,7 +1177,7 @@ public class BigQueryServicesImpl implements BigQueryServices {
// If adding the next row will push the request above BQ row limits, or
// if the current batch of elements is larger than the targeted request size,
// we immediately go and issue the data insertion.
- if (dataSize + nextRowSize >= MAX_BQ_ROW_PAYLOAD_BYTES
+ if (dataSize + nextRowSize >= MAX_BQ_ROW_PAYLOAD
|| dataSize >= maxRowBatchSize
|| rows.size() + 1 > maxRowsPerBatch) {
// If the row does not fit into the insert buffer, then we take the current buffer,
@@ -1356,48 +1317,6 @@ public class BigQueryServicesImpl implements BigQueryServices {
}
}
- /**
- * Validates a {@link TableRow} for logging, comparing the provided row against a BigQuery
- * schema. The formatted string shows the field names in the row indicating any mismatches
- * unknown entries.
- *
- * <p>For example, a {@link TableRow} with a "names" field, where the schema expects
- * "name" would return "names".
- *
- * <pre>{@code {'name': java.lang.String}</pre>
- *
- * <p>If a field exists in the row but not in the schema,
- * "Unknown fields" is prefixed to the log.</p>
- *
- * @param row The {@link TableRow} to validate.
- * @param tableSchema The {@link TableSchema} to check against.
- * @return A string representation of the row, indicating any schema mismatches.
- */
- private String validateRowSchema(TableRow row, TableSchema tableSchema) {
- // Creates bqSchemaFields containing field names
- Set<String> bqSchemaFields =
- tableSchema.getFieldsList().stream().map(f -> f.getName()).collect(Collectors.toSet());
-
- // Validate
- String rowDetails =
- row.keySet().stream()
- .map(
- fieldName -> {
- if (!bqSchemaFields.contains(fieldName)) {
- return fieldName;
- }
- return "";
- })
- .filter(s -> !s.isEmpty())
- .collect(Collectors.joining(", ", "{Unknown fields: ", "}"));
-
- // Shorten row details if too long for human readability
- if (rowDetails.length() > 1024) {
- rowDetails = rowDetails.substring(0, 1024) + "...}";
- }
- return rowDetails;
- }
-
@Override
public <T> long insertAll(
TableReference ref,
diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImplTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImplTest.java
index 3902fb1fca3..01454a50b25 100644
--- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImplTest.java
+++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImplTest.java
@@ -1119,11 +1119,6 @@ public class BigQueryServicesImplTest {
public void testInsertWithinRequestByteSizeLimitsErrorsOut() throws Exception {
TableReference ref =
new TableReference().setProjectId("project").setDatasetId("dataset").setTableId("tablersl");
- TableSchema schema =
- new TableSchema()
- .setFields(ImmutableList.of(new TableFieldSchema().setName("rows").setType("STRING")));
- Table testTable = new Table().setTableReference(ref).setSchema(schema);
-
List<FailsafeValueInSingleWindow<TableRow, TableRow>> rows =
ImmutableList.of(
wrapValue(new TableRow().set("row", Strings.repeat("abcdefghi", 1024 * 1025))),
@@ -1138,81 +1133,11 @@ public class BigQueryServicesImplTest {
when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
when(response.getStatusCode()).thenReturn(200);
when(response.getContent()).thenReturn(toStream(allRowsSucceeded));
- when(response.getContent()).thenReturn(toStream(testTable));
},
response -> {
when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
when(response.getStatusCode()).thenReturn(200);
when(response.getContent()).thenReturn(toStream(allRowsSucceeded));
- when(response.getContent()).thenReturn(toStream(testTable));
- });
-
- DatasetServiceImpl dataService =
- new DatasetServiceImpl(
- bigquery, PipelineOptionsFactory.fromArgs("--maxStreamingBatchSize=15").create());
- List<ValueInSingleWindow<TableRow>> failedInserts = Lists.newArrayList();
- List<ValueInSingleWindow<TableRow>> successfulRows = Lists.newArrayList();
- RuntimeException e =
- assertThrows(
- RuntimeException.class,
- () ->
- dataService.<TableRow>insertAll(
- ref,
- rows,
- insertIds,
- BackOffAdapter.toGcpBackOff(TEST_BACKOFF.backoff()),
- TEST_BACKOFF,
- new MockSleeper(),
- InsertRetryPolicy.alwaysRetry(),
- failedInserts,
- ErrorContainer.TABLE_ROW_ERROR_CONTAINER,
- false,
- false,
- false,
- successfulRows));
-
- assertThat(e.getMessage(), containsString("exceeding the BigQueryIO limit"));
- }
-
- /**
- * Tests that {@link DatasetServiceImpl#insertAll} does not go over limit of rows per request and
- * schema difference check.
- */
- @SuppressWarnings("InlineMeInliner") // inline `Strings.repeat()` - Java 11+ API only
- @Test
- public void testInsertWithinRequestByteSizeLimitsWithBadSchemaErrorsOut() throws Exception {
- TableReference ref =
- new TableReference().setProjectId("project").setDatasetId("dataset").setTableId("tablersl");
-
- TableSchema schema =
- new TableSchema()
- .setFields(ImmutableList.of(new TableFieldSchema().setName("row").setType("STRING")));
- Table testTable = new Table().setTableReference(ref).setSchema(schema);
-
- List<FailsafeValueInSingleWindow<TableRow, TableRow>> rows =
- ImmutableList.of(
- wrapValue(
- new TableRow()
- .set("row", Strings.repeat("abcdefghi", 1024 * 1025))
- .set("badField", "goodValue")),
- wrapValue(new TableRow().set("row", "a")),
- wrapValue(new TableRow().set("row", "b")));
- List<String> insertIds = ImmutableList.of("a", "b", "c");
-
- final TableDataInsertAllResponse allRowsSucceeded = new TableDataInsertAllResponse();
-
- setupMockResponses(
- response -> {
- when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
- when(response.getStatusCode()).thenReturn(200);
- when(response.getContent()).thenReturn(toStream(allRowsSucceeded));
- when(response.getContent()).thenReturn(toStream(testTable));
- },
- response -> {
- when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
- when(response.getStatusCode()).thenReturn(200);
- when(response.getContent()).thenReturn(toStream(allRowsSucceeded));
- when(response.getContent()).thenReturn(toStream(testTable));
});
DatasetServiceImpl dataService =
@@ -1239,10 +1164,7 @@ public class BigQueryServicesImplTest {
false,
successfulRows));
- assertThat(e.getMessage(), containsString("exceeding the BigQueryIO limit"));
- assertThat(
- e.getMessage(),
- containsString("Problematic row had extra schema fields: {Unknown fields: badField}"));
+ assertThat(e.getMessage(), containsString("exceeded BigQueryIO limit of 9MB."));
}
@SuppressWarnings("InlineMeInliner") // inline `Strings.repeat()` - Java 11+ API only