You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2026/07/29 10:32:43 UTC
[L] Change in asterixdb[master]: MB-70477: support max-file-size for COPY TO
From Rithwik Koul <ri...@couchbase.com>:
Rithwik Koul has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21471?usp=email )
Change subject: MB-70477: support max-file-size for COPY TO
......................................................................
MB-70477: support max-file-size for COPY TO
Add a "max-file-size" WITH option that bounds the size of each file
produced by COPY TO, complementing the existing "max-objects-per-file"
row cap. Without a size cap nothing bounds a single object, and an S3
multipart upload fails only once it exceeds part 10,001 (~78 GiB), i.e.
at the very end of a long-running query.
The cap is evaluated in ExternalFileWriter alongside the row cap, using
the same check-before-write placement so the last tuple does not trigger
a rollover that would publish an empty object.
Reporting the current file size requires two sources. Cloud, HDFS and
local writers know how many bytes the backend has accepted, but a
block-oriented printer buffers a whole row group in memory and hands
nothing to the stream until that group is flushed, so the backend
counter alone stays at the file magic for the entire file whenever the
row group is larger than the cap. IExternalPrinter therefore gains a
default getBytesWritten() returning 0, overridden only by
ParquetExternalFilePrinter, and each file writer returns the maximum of
the two terms. The maximum, not the sum: ParquetWriter.getDataSize()
already includes previously flushed row groups.
The value is approximate by design. It reflects bytes after compression
and excludes bytes produced only on close, such as the Parquet footer.
It is also not monotonic for Parquet, since a flush re-accounts the open
row group from encoded to compressed size, so it is safe for a threshold
test but must not be used to compute deltas.
Validation accepts 5MB..5TB with -1 for unlimited, matching the S3
minimum part size.
Change-Id: I44e6777175c507b1c289fd7cb8985e23331fdd49
---
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.01.ddl.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.02.update.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.03.update.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.04.ddl.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.05.query.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.01.ddl.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.02.update.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.03.update.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.04.update.sqlpp
A asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/max-file-size/max-file-size.05.adm
M asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_azure_blob_storage.xml
M asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
M asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp_hdfs.xml
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetExternalWriterFactory.java
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetSinkExternalWriterFactory.java
M asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/AbstractCloudExternalFileWriter.java
M asterixdb/asterix-doc/src/main/markdown/sqlpp/7_ddl_dml.md
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/WriterValidationUtil.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/HDFSExternalFileWriter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/LocalFSExternalFileWriter.java
M asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/ParquetExternalFilePrinter.java
A asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/WriterMaxFileSizeValidationTest.java
A asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/writer/LocalFSMaxFileSizeDriverTest.java
M asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriter.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriterFactory.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalFileWriter.java
M asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalPrinter.java
A asterixdb/asterix-runtime/src/test/java/org/apache/asterix/runtime/writer/ExternalFileWriterMaxFileSizeTest.java
30 files changed, 931 insertions(+), 15 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/71/21471/1
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.01.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.01.ddl.sqlpp
new file mode 100644
index 0000000..68b9812
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.01.ddl.sqlpp
@@ -0,0 +1,28 @@
+/*
+ * 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.
+ */
+
+DROP DATAVERSE test if exists;
+CREATE DATAVERSE test;
+USE test;
+
+CREATE TYPE ColumnType1 AS {
+ id: integer
+ };
+
+CREATE COLLECTION snapshots(ColumnType1) PRIMARY KEY id;
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.02.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.02.update.sqlpp
new file mode 100644
index 0000000..c10705d
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.02.update.sqlpp
@@ -0,0 +1,35 @@
+/*
+ * 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.
+ */
+
+/*
+ * Each object is roughly 3KB, so 4000 objects write well past the 5MB limit used in the COPY TO below
+ * and the output has to be split over several files.
+ */
+
+USE test;
+
+INSERT INTO snapshots(
+ SELECT VALUE {
+ "id": intVal,
+ "padding_a": repeat("a", 1000),
+ "padding_b": repeat("b", 1000),
+ "padding_c": repeat("c", 1000)
+ }
+ FROM range(1, 4000) intVal
+);
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.03.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.03.update.sqlpp
new file mode 100644
index 0000000..84d20da
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.03.update.sqlpp
@@ -0,0 +1,32 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY (
+ SELECT c.* FROM snapshots c ORDER BY c.id
+ ) toWriter
+TO %adapter%
+PATH (%pathprefix% "copy-to-result", "max-file-size")
+WITH {
+ %template_colons%,
+ %additionalProperties%
+ "format":"json",
+ "max-file-size": "5MB"
+ };
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.04.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.04.ddl.sqlpp
new file mode 100644
index 0000000..17f7622
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.04.ddl.sqlpp
@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+CREATE TYPE ColumnType2 AS {
+ };
+
+CREATE EXTERNAL DATASET TestDataset(ColumnType2) USING %adapter%
+(
+ %template%,
+ %additional_Properties%
+ ("definition"="%path_prefix%copy-to-result/max-file-size/"),
+ ("include"="*.json"),
+ ("requireVersionChangeDetection"="false"),
+ ("format" = "json")
+);
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.05.query.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.05.query.sqlpp
new file mode 100644
index 0000000..a6965e6
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/max-file-size/max-file-size.05.query.sqlpp
@@ -0,0 +1,32 @@
+/*
+ * 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.
+ */
+
+/*
+ * Splitting the output over several files must not lose or duplicate any object, so reading all of them
+ * back has to reproduce the source collection exactly.
+ */
+
+USE test;
+
+SELECT VALUE {
+ "objects": count(c.id),
+ "distinct_ids": count(DISTINCT c.id),
+ "id_sum": sum(c.id)
+}
+FROM TestDataset c;
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.01.ddl.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.01.ddl.sqlpp
new file mode 100644
index 0000000..6f4cc98
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.01.ddl.sqlpp
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+DROP DATAVERSE test IF EXISTS;
+CREATE DATAVERSE test;
+
+USE test;
+
+CREATE TYPE OpenType AS {
+};
+
+CREATE EXTERNAL DATASET Customer(OpenType) USING %adapter% (
+ %template%,
+ %additional_Properties%
+ ("definition"="%path_prefix%external-filter/car/{company:string}/customer/{customer_id:int}"),
+ ("embed-filter-values" = "false"),
+ ("format"="json")
+);
+
+
+
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.02.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.02.update.sqlpp
new file mode 100644
index 0000000..7146a1b
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.02.update.sqlpp
@@ -0,0 +1,30 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY Customer c
+TO %adapter%
+PATH (%pathprefix% "copy-to-result-with-over")
+WITH {
+ %template_colons%,
+ %additionalProperties%
+ "format":"json",
+ "max-file-size": "1MB"
+}
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.03.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.03.update.sqlpp
new file mode 100644
index 0000000..101453f
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.03.update.sqlpp
@@ -0,0 +1,30 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY Customer c
+TO %adapter%
+PATH (%pathprefix% "copy-to-result-with-over")
+WITH {
+ %template_colons%,
+ %additionalProperties%
+ "format":"json",
+ "max-file-size": "random"
+}
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.04.update.sqlpp b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.04.update.sqlpp
new file mode 100644
index 0000000..9f11603
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/copy-to/negative/bad-max-file-size/max-file-size.04.update.sqlpp
@@ -0,0 +1,30 @@
+/*
+ * 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.
+ */
+
+USE test;
+
+COPY Customer c
+TO %adapter%
+PATH (%pathprefix% "copy-to-result-with-over")
+WITH {
+ %template_colons%,
+ %additionalProperties%
+ "format":"json",
+ "max-file-size": "6TB"
+}
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/max-file-size/max-file-size.05.adm b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/max-file-size/max-file-size.05.adm
new file mode 100644
index 0000000..2992765
--- /dev/null
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/results/copy-to/max-file-size/max-file-size.05.adm
@@ -0,0 +1 @@
+{ "objects": 4000, "distinct_ids": 4000, "id_sum": 8002000 }
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_azure_blob_storage.xml b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_azure_blob_storage.xml
index 993372e..4c65368 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_azure_blob_storage.xml
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_azure_blob_storage.xml
@@ -274,6 +274,19 @@
</compilation-unit>
</test-case>
<test-case FilePath="copy-to/negative">
+ <compilation-unit name="bad-max-file-size">
+ <placeholder name="adapter" value="AZUREBLOB" />
+ <placeholder name="pathprefix" value="" />
+ <placeholder name="path_prefix" value="" />
+ <placeholder name="additionalProperties" value='"container":"playground",' />
+ <placeholder name="additional_Properties" value='("container"="playground"),' />
+ <output-dir compare="Text">bad-max-file-size</output-dir>
+ <expected-error>Minimum value allowed for 'max-file-size' is 5 MiB. Found 1MB</expected-error>
+ <expected-error>ASX1206: Storage units expected for the field 'max-file-size' (e.g., 0.1KB, 100kb, 1mb, 3MB, 8.5GB ...). Provided 'random'</expected-error>
+ <expected-error>Maximum value allowed for 'max-file-size' is 5 TiB. Found 6TB</expected-error>
+ </compilation-unit>
+ </test-case>
+ <test-case FilePath="copy-to/negative">
<compilation-unit name="csv-error-checks">
<placeholder name="adapter" value="AZUREBLOB" />
<placeholder name="pathprefix" value="" />
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
index d19d5fc..c12c157 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_external_dataset_s3.xml
@@ -190,6 +190,16 @@
</compilation-unit>
</test-case>
<test-case FilePath="copy-to">
+ <compilation-unit name="max-file-size">
+ <placeholder name="adapter" value="S3" />
+ <placeholder name="pathprefix" value="" />
+ <placeholder name="path_prefix" value="" />
+ <placeholder name="additionalProperties" value='"container":"playground",' />
+ <placeholder name="additional_Properties" value='("container"="playground"),' />
+ <output-dir compare="Text">max-file-size</output-dir>
+ </compilation-unit>
+ </test-case>
+ <test-case FilePath="copy-to">
<compilation-unit name="parquet-field-names">
<placeholder name="adapter" value="S3" />
<placeholder name="pathprefix" value="" />
@@ -373,6 +383,19 @@
</compilation-unit>
</test-case>
<test-case FilePath="copy-to/negative">
+ <compilation-unit name="bad-max-file-size">
+ <placeholder name="adapter" value="S3" />
+ <placeholder name="pathprefix" value="" />
+ <placeholder name="path_prefix" value="" />
+ <placeholder name="additionalProperties" value='"container":"playground",' />
+ <placeholder name="additional_Properties" value='("container"="playground"),' />
+ <output-dir compare="Text">bad-max-file-size</output-dir>
+ <expected-error>Minimum value allowed for 'max-file-size' is 5 MiB. Found 1MB</expected-error>
+ <expected-error>ASX1206: Storage units expected for the field 'max-file-size' (e.g., 0.1KB, 100kb, 1mb, 3MB, 8.5GB ...). Provided 'random'</expected-error>
+ <expected-error>Maximum value allowed for 'max-file-size' is 5 TiB. Found 6TB</expected-error>
+ </compilation-unit>
+ </test-case>
+ <test-case FilePath="copy-to/negative">
<compilation-unit name="csv-error-checks">
<placeholder name="adapter" value="S3" />
<placeholder name="pathprefix" value="" />
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp_hdfs.xml b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp_hdfs.xml
index c0635af..f50169c 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp_hdfs.xml
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/testsuite_sqlpp_hdfs.xml
@@ -293,6 +293,20 @@
</compilation-unit>
</test-case>
<test-case FilePath="copy-to/negative">
+ <compilation-unit name="bad-max-file-size">
+ <placeholder name="adapter" value="HDFS" />
+ <placeholder name="pathprefix" value='"/playground", ' />
+ <placeholder name="path_prefix" value="/playground/" />
+ <placeholder name="additionalProperties" value="" />
+ <placeholder name="additional_Properties" value='("input-format" = "text-input-format"),' />
+ <output-dir compare="Text">bad-max-file-size</output-dir>
+ <expected-error>Minimum value allowed for 'max-file-size' is 5 MiB. Found 1MB</expected-error>
+ <expected-error>ASX1206: Storage units expected for the field 'max-file-size' (e.g., 0.1KB, 100kb, 1mb, 3MB, 8.5GB ...). Provided 'random'</expected-error>
+ <expected-error>Maximum value allowed for 'max-file-size' is 5 TiB. Found 6TB</expected-error>
+ <source-location>false</source-location>
+ </compilation-unit>
+ </test-case>
+ <test-case FilePath="copy-to/negative">
<compilation-unit name="csv-error-checks">
<placeholder name="adapter" value="HDFS" />
<placeholder name="pathprefix" value='"/playground", ' />
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetExternalWriterFactory.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetExternalWriterFactory.java
index 5d9ab7f..f3de8a8 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetExternalWriterFactory.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetExternalWriterFactory.java
@@ -38,15 +38,17 @@
private static final long serialVersionUID = 8971234908711236L;
private final IExternalFileWriterFactory writerFactory;
private final int maxResult;
+ private final long maxFileSize;
private final ParquetExternalFilePrinterFactory printerFactory;
private final IPathResolver resolver;
private final IHyracksTaskContext ctx;
public ParquetExternalWriterFactory(IHyracksTaskContext ctx, IExternalFileWriterFactory writerFactory,
- int maxResult, ParquetExternalFilePrinterFactory printerFactory, IPathResolver resolver) {
+ int maxResult, long maxFileSize, ParquetExternalFilePrinterFactory printerFactory, IPathResolver resolver) {
this.ctx = ctx;
this.writerFactory = writerFactory;
this.maxResult = maxResult;
+ this.maxFileSize = maxFileSize;
this.printerFactory = printerFactory;
this.resolver = resolver;
}
@@ -55,7 +57,7 @@
MessageType schema = generateSchema(schemaNode);
printerFactory.setParquetSchema(schema);
IExternalFileWriter writer = writerFactory.createWriter(ctx, printerFactory);
- return new ExternalFileWriter(resolver, writer, maxResult);
+ return new ExternalFileWriter(resolver, writer, maxResult, maxFileSize);
}
}
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetSinkExternalWriterFactory.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetSinkExternalWriterFactory.java
index 3a276a2..04dc860 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetSinkExternalWriterFactory.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/parquet/ParquetSinkExternalWriterFactory.java
@@ -39,12 +39,13 @@
private final IAType sourceType;
private final IExternalFileWriterFactory writerFactory;
private final int maxResult;
+ private final long maxFileSize;
private final ParquetExternalFilePrinterFactoryProvider printerFactoryProvider;
private final IPathResolverFactory pathResolverFactory;
public ParquetSinkExternalWriterFactory(WriterPartitionerFactory partitionerFactory,
RecordDescriptor inputRecordDesc, int sourceColumn, IAType sourceType, int maxSchemas,
- IExternalFileWriterFactory writerFactory, int maxResult,
+ IExternalFileWriterFactory writerFactory, int maxResult, long maxFileSize,
ParquetExternalFilePrinterFactoryProvider printerFactoryProvider,
IPathResolverFactory pathResolverFactory) {
this.partitionerFactory = partitionerFactory;
@@ -54,6 +55,7 @@
this.maxSchemas = maxSchemas;
this.writerFactory = writerFactory;
this.maxResult = maxResult;
+ this.maxFileSize = maxFileSize;
this.printerFactoryProvider = printerFactoryProvider;
this.pathResolverFactory = pathResolverFactory;
}
@@ -63,7 +65,7 @@
ParquetExternalFilePrinterFactory printerFactory =
(ParquetExternalFilePrinterFactory) printerFactoryProvider.createPrinterFactory();
ParquetExternalWriterFactory parquetExternalWriterFactory = new ParquetExternalWriterFactory(ctx, writerFactory,
- maxResult, printerFactory, pathResolverFactory.createResolver(ctx));
+ maxResult, maxFileSize, printerFactory, pathResolverFactory.createResolver(ctx));
ParquetSinkExternalWriterRuntime runtime =
new ParquetSinkExternalWriterRuntime(sourceColumn, partitionerFactory.createPartitioner(),
inputRecordDesc, parquetExternalWriterFactory, sourceType, maxSchemas);
diff --git a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/AbstractCloudExternalFileWriter.java b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/AbstractCloudExternalFileWriter.java
index e389c03..d7a5f10 100644
--- a/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/AbstractCloudExternalFileWriter.java
+++ b/asterixdb/asterix-cloud/src/main/java/org/apache/asterix/cloud/writer/AbstractCloudExternalFileWriter.java
@@ -107,6 +107,14 @@
}
@Override
+ public final long getBytesWritten() {
+ // must not be called from newFile(): the new writer is installed before the previous stream is
+ // finished, so it would report the new, empty file
+ long acceptedBytes = cloudWriter == null ? 0 : cloudWriter.position();
+ return Math.max(acceptedBytes, printer.getBytesWritten());
+ }
+
+ @Override
public final void abort() throws HyracksDataException {
try {
if (cloudWriter != null) {
diff --git a/asterixdb/asterix-doc/src/main/markdown/sqlpp/7_ddl_dml.md b/asterixdb/asterix-doc/src/main/markdown/sqlpp/7_ddl_dml.md
index 4f7174c..c149d5a 100644
--- a/asterixdb/asterix-doc/src/main/markdown/sqlpp/7_ddl_dml.md
+++ b/asterixdb/asterix-doc/src/main/markdown/sqlpp/7_ddl_dml.md
@@ -777,6 +777,7 @@
"format" : "json",
"compression": "gzip",
"max-objects-per-file": 1000,
+ "max-file-size": "128MB",
"container": "myBucket",
"accessKeyId": "<access-key>",
"secretAccessKey": "<secret-key>",
@@ -784,5 +785,13 @@
};
This query will be exported as partitions into a set of folders, with one folder for each value of `battery_status`.
-Each partition itself will also be sorted by the `name` field, and compressed with `gzip` and divided into files of 100
+Each partition itself will also be sorted by the `name` field, and compressed with `gzip` and divided into files of 1000
objects or fewer per file.
+
+A file is also started afresh once it reaches `max-file-size`, so whichever of the two limits is reached first ends the
+current file. `max-file-size` accepts a storage unit string such as `5MB` or `1GB`, has a minimum of `5MB`, and is
+unbounded when it is not specified. It is measured on the bytes written to the target after compression and is
+approximate: an object is never split across files, so the object that reaches the limit takes the file past it. For
+`parquet` the size is only observable at row group boundaries, so `max-file-size` should be set comfortably above
+`row-group-size`. The largest file the target itself accepts is generally lower than the maximum accepted here, as it
+depends on the target's multipart upload limits.
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
index c8fc177..76831d6 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/ExternalDataConstants.java
@@ -373,10 +373,16 @@
* Writer Constants
*/
public static final String KEY_WRITER_MAX_RESULT = "max-objects-per-file";
+ public static final String KEY_WRITER_MAX_FILE_SIZE = "max-file-size";
public static final String KEY_VALIDATE_WRITE_PERMISSION = "validate-write-permission";
public static final String KEY_WRITER_COMPRESSION = "compression";
public static final int WRITER_MAX_RESULT_DEFAULT = 10000;
public static final int WRITER_MAX_RESULT_MINIMUM = 1000;
+ public static final long WRITER_MAX_FILE_SIZE_UNLIMITED = -1L;
+ public static final long WRITER_MAX_FILE_SIZE_MINIMUM =
+ StorageUtil.getLongSizeInBytes(5, StorageUtil.StorageUnit.MEGABYTE);
+ public static final long WRITER_MAX_FILE_SIZE_MAXIMUM =
+ StorageUtil.getLongSizeInBytes(5, StorageUtil.StorageUnit.TERABYTE);
public static final Set<String> WRITER_SUPPORTED_FORMATS;
public static final Set<String> WRITER_SUPPORTED_ADAPTERS;
public static final Set<String> TEXTUAL_WRITER_SUPPORTED_COMPRESSION;
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/WriterValidationUtil.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/WriterValidationUtil.java
index 92afd40..0bb2bee 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/WriterValidationUtil.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/util/WriterValidationUtil.java
@@ -31,10 +31,13 @@
import static org.apache.asterix.external.util.ExternalDataConstants.FORMAT_PARQUET;
import static org.apache.asterix.external.util.ExternalDataConstants.KEY_PARQUET_PAGE_SIZE;
import static org.apache.asterix.external.util.ExternalDataConstants.KEY_PARQUET_ROW_GROUP_SIZE;
+import static org.apache.asterix.external.util.ExternalDataConstants.KEY_WRITER_MAX_FILE_SIZE;
import static org.apache.asterix.external.util.ExternalDataConstants.KEY_WRITER_MAX_RESULT;
import static org.apache.asterix.external.util.ExternalDataConstants.PARQUET_MAX_SCHEMAS_KEY;
import static org.apache.asterix.external.util.ExternalDataConstants.PARQUET_MAX_SCHEMAS_MAX_VALUE;
import static org.apache.asterix.external.util.ExternalDataConstants.PARQUET_WRITER_VERSION_KEY;
+import static org.apache.asterix.external.util.ExternalDataConstants.WRITER_MAX_FILE_SIZE_MAXIMUM;
+import static org.apache.asterix.external.util.ExternalDataConstants.WRITER_MAX_FILE_SIZE_MINIMUM;
import static org.apache.asterix.external.util.ExternalDataConstants.WRITER_MAX_RESULT_MINIMUM;
import java.util.List;
@@ -57,6 +60,7 @@
validateAdapter(adapter, supportedAdapters, sourceLocation);
validateFormat(configuration, sourceLocation);
validateMaxResult(configuration, sourceLocation);
+ validateMaxFileSize(configuration);
}
private static void validateQuote(Map<String, String> configuration, SourceLocation sourceLocation)
@@ -183,6 +187,29 @@
}
}
+ private static void validateMaxFileSize(Map<String, String> configuration) throws CompilationException {
+ String maxFileSize = configuration.get(KEY_WRITER_MAX_FILE_SIZE);
+ if (maxFileSize == null) {
+ return;
+ }
+
+ long value;
+ try {
+ value = StorageUtil.getByteValue(maxFileSize);
+ } catch (IllegalArgumentException e) {
+ throw CompilationException.create(ErrorCode.ILLEGAL_SIZE_PROVIDED, KEY_WRITER_MAX_FILE_SIZE, maxFileSize);
+ }
+
+ if (value < WRITER_MAX_FILE_SIZE_MINIMUM) {
+ throw new CompilationException(MINIMUM_VALUE_ALLOWED_FOR_PARAM, KEY_WRITER_MAX_FILE_SIZE,
+ StorageUtil.toHumanReadableSize(WRITER_MAX_FILE_SIZE_MINIMUM), maxFileSize);
+ }
+ if (value > WRITER_MAX_FILE_SIZE_MAXIMUM) {
+ throw new CompilationException(MAXIMUM_VALUE_ALLOWED_FOR_PARAM, KEY_WRITER_MAX_FILE_SIZE,
+ StorageUtil.toHumanReadableSize(WRITER_MAX_FILE_SIZE_MAXIMUM), maxFileSize);
+ }
+ }
+
private static void validateMaxParquetSchemas(Map<String, String> configuration, SourceLocation sourceLocation)
throws CompilationException {
String maxResult = configuration.get(PARQUET_MAX_SCHEMAS_KEY);
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/HDFSExternalFileWriter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/HDFSExternalFileWriter.java
index 510c8e4..e8f7e63 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/HDFSExternalFileWriter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/HDFSExternalFileWriter.java
@@ -109,6 +109,12 @@
}
@Override
+ public long getBytesWritten() {
+ long writtenBytes = outputStream == null ? 0 : outputStream.getPos();
+ return Math.max(writtenBytes, printer.getBytesWritten());
+ }
+
+ @Override
public void abort() throws HyracksDataException {
try {
printer.close();
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/LocalFSExternalFileWriter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/LocalFSExternalFileWriter.java
index 906e272..e1d0bd4 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/LocalFSExternalFileWriter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/LocalFSExternalFileWriter.java
@@ -36,6 +36,7 @@
private final IExternalPrinter printer;
private final ILocalFSValidator validator;
private final SourceLocation pathSourceLocation;
+ private FileOutputStream currentFileStream;
LocalFSExternalFileWriter(IExternalPrinter printer, ILocalFSValidator validator,
SourceLocation pathSourceLocation) {
@@ -64,7 +65,8 @@
if (!currentFile.createNewFile()) {
throw RuntimeDataException.create(ErrorCode.COULD_NOT_CREATE_FILE, currentFile.getAbsolutePath());
}
- printer.newStream(new BufferedOutputStream(new FileOutputStream(currentFile)));
+ currentFileStream = new FileOutputStream(currentFile);
+ printer.newStream(new BufferedOutputStream(currentFileStream));
} catch (IOException e) {
throw HyracksDataException.create(e);
}
@@ -77,6 +79,17 @@
}
@Override
+ public long getBytesWritten() throws HyracksDataException {
+ try {
+ // lags behind by whatever the buffered stream has not flushed yet
+ long writtenBytes = currentFileStream == null ? 0 : currentFileStream.getChannel().position();
+ return Math.max(writtenBytes, printer.getBytesWritten());
+ } catch (IOException e) {
+ throw HyracksDataException.create(e);
+ }
+ }
+
+ @Override
public void abort() throws HyracksDataException {
printer.close();
}
diff --git a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/ParquetExternalFilePrinter.java b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/ParquetExternalFilePrinter.java
index 3ef41b9..becfe30 100644
--- a/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/ParquetExternalFilePrinter.java
+++ b/asterixdb/asterix-external-data/src/main/java/org/apache/asterix/external/writer/printer/ParquetExternalFilePrinter.java
@@ -82,6 +82,13 @@
}
@Override
+ public long getBytesWritten() {
+ // flushed row groups are counted as compressed bytes, the open row group as encoded bytes; the
+ // footer is only written on close and is therefore not counted
+ return writer == null ? 0 : writer.getDataSize();
+ }
+
+ @Override
public void print(IValueReference value) throws HyracksDataException {
try {
this.writer.write(value);
diff --git a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/WriterMaxFileSizeValidationTest.java b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/WriterMaxFileSizeValidationTest.java
new file mode 100644
index 0000000..13966af
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/util/WriterMaxFileSizeValidationTest.java
@@ -0,0 +1,108 @@
+/*
+ * 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.asterix.external.util;
+
+import static org.apache.asterix.external.util.ExternalDataConstants.KEY_WRITER_MAX_FILE_SIZE;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.asterix.common.exceptions.CompilationException;
+import org.apache.asterix.common.exceptions.ErrorCode;
+import org.junit.Test;
+
+public class WriterMaxFileSizeValidationTest {
+
+ private static final String ADAPTER = ExternalDataConstants.KEY_ADAPTER_NAME_AWS_S3;
+
+ @Test
+ public void testAbsentIsAccepted() throws CompilationException {
+ validate(null);
+ }
+
+ @Test
+ public void testValidSizesAreAccepted() throws CompilationException {
+ validate("5MB");
+ validate("128MB");
+ validate("1GB");
+ validate("8.5GB");
+ validate("5TB");
+ validate("10485760");
+ }
+
+ @Test
+ public void testUnparsableSizeIsRejected() {
+ assertRejected("random", ErrorCode.ILLEGAL_SIZE_PROVIDED);
+ assertRejected("", ErrorCode.ILLEGAL_SIZE_PROVIDED);
+ assertRejected("10XB", ErrorCode.ILLEGAL_SIZE_PROVIDED);
+ }
+
+ @Test
+ public void testSizeBelowMinimumIsRejected() {
+ assertRejected("1MB", ErrorCode.MINIMUM_VALUE_ALLOWED_FOR_PARAM);
+ assertRejected("1024", ErrorCode.MINIMUM_VALUE_ALLOWED_FOR_PARAM);
+ assertRejected("0", ErrorCode.MINIMUM_VALUE_ALLOWED_FOR_PARAM);
+ }
+
+ @Test
+ public void testNegativeSizeIsRejected() {
+ // the size parser accepts a leading minus, so the minimum check is what rules these out
+ assertRejected("-1", ErrorCode.MINIMUM_VALUE_ALLOWED_FOR_PARAM);
+ assertRejected("-10MB", ErrorCode.MINIMUM_VALUE_ALLOWED_FOR_PARAM);
+ }
+
+ @Test
+ public void testSizeAboveMaximumIsRejected() {
+ assertRejected("6TB", ErrorCode.MAXIMUM_VALUE_ALLOWED_FOR_PARAM);
+ assertRejected("1PB", ErrorCode.MAXIMUM_VALUE_ALLOWED_FOR_PARAM);
+ }
+
+ @Test
+ public void testRejectionNamesTheOption() {
+ try {
+ validate("random");
+ fail("expected the invalid size to be rejected");
+ } catch (CompilationException e) {
+ assertTrue("error message does not name the option: " + e.getMessage(),
+ e.getMessage().contains(KEY_WRITER_MAX_FILE_SIZE));
+ }
+ }
+
+ private static void assertRejected(String maxFileSize, ErrorCode expected) {
+ try {
+ validate(maxFileSize);
+ fail("expected '" + maxFileSize + "' to be rejected");
+ } catch (CompilationException e) {
+ assertEquals("unexpected error for '" + maxFileSize + "'", expected.intValue(), e.getErrorCode());
+ }
+ }
+
+ private static void validate(String maxFileSize) throws CompilationException {
+ Map<String, String> configuration = new HashMap<>();
+ configuration.put(ExternalDataConstants.KEY_FORMAT, ExternalDataConstants.FORMAT_JSON_LOWER_CASE);
+ if (maxFileSize != null) {
+ configuration.put(KEY_WRITER_MAX_FILE_SIZE, maxFileSize);
+ }
+ WriterValidationUtil.validateWriterConfiguration(ADAPTER, ExternalDataConstants.WRITER_SUPPORTED_ADAPTERS,
+ configuration, null);
+ }
+}
diff --git a/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/writer/LocalFSMaxFileSizeDriverTest.java b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/writer/LocalFSMaxFileSizeDriverTest.java
new file mode 100644
index 0000000..3822765
--- /dev/null
+++ b/asterixdb/asterix-external-data/src/test/java/org/apache/asterix/external/writer/LocalFSMaxFileSizeDriverTest.java
@@ -0,0 +1,167 @@
+/*
+ * 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.asterix.external.writer;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.Arrays;
+
+import org.apache.asterix.runtime.writer.ExternalFileWriter;
+import org.apache.asterix.runtime.writer.IExternalPrinter;
+import org.apache.asterix.runtime.writer.StaticPathResolver;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.api.IValueReference;
+import org.junit.Test;
+
+/**
+ * Drives the real writer stack against the local filesystem so the produced files can be inspected.
+ */
+public class LocalFSMaxFileSizeDriverTest {
+ private static final long MAX_FILE_SIZE = 5 * 1024 * 1024L;
+ private static final int NO_MAX_RESULT = Integer.MAX_VALUE;
+ private static final int OBJECT_SIZE = 3000;
+ private static final int OBJECTS = 20000;
+ private static final String OUTPUT_DIRECTORY =
+ "/private/tmp/claude-502/-Users-rithwik-koul-Desktop-Couchbase-cb1/2bfb98a5-4716-4670-b3ef-09b4e460ef92/scratchpad/copy-to-out";
+
+ @Test
+ public void writeAndListFiles() throws Exception {
+ File directory = new File(OUTPUT_DIRECTORY);
+ if (directory.exists()) {
+ for (File stale : directory.listFiles()) {
+ Files.delete(stale.toPath());
+ }
+ }
+
+ RawBytePrinter printer = new RawBytePrinter();
+ LocalFSExternalFileWriter fileWriter = new LocalFSExternalFileWriter(printer, (dir, location) -> {
+ }, null);
+ ExternalFileWriter writer =
+ new ExternalFileWriter(new StaticPathResolver("json", File.separatorChar, 0, OUTPUT_DIRECTORY),
+ fileWriter, NO_MAX_RESULT, MAX_FILE_SIZE);
+
+ writer.open();
+ writer.initNewPartition(null);
+ for (int id = 1; id <= OBJECTS; id++) {
+ writer.write(object(id));
+ }
+ writer.close();
+
+ File[] produced = directory.listFiles();
+ Arrays.sort(produced);
+ long total = 0;
+ System.out.println("=== BEGIN LISTING ===");
+ System.out.println("maxFileSize=" + MAX_FILE_SIZE + " objectSize=" + OBJECT_SIZE + " objects=" + OBJECTS);
+ for (File file : produced) {
+ long length = file.length();
+ total += length;
+ System.out.println(String.format("%-34s %10d bytes %7.3f objects overshoot %+d", file.getName(), length,
+ length / (double) OBJECT_SIZE, length - MAX_FILE_SIZE));
+ }
+ System.out.println(
+ "files=" + produced.length + " totalBytes=" + total + " expected=" + (OBJECTS * (long) OBJECT_SIZE));
+ System.out.println("sizeQueries=" + printer.sizeQueries);
+ System.out.println("=== END LISTING ===");
+ }
+
+ private static IValueReference object(int id) {
+ byte[] prefix = ("{\"id\":" + id + ",\"padding\":\"").getBytes(StandardCharsets.UTF_8);
+ byte[] suffix = "\"}\n".getBytes(StandardCharsets.UTF_8);
+ byte[] bytes = new byte[OBJECT_SIZE];
+ Arrays.fill(bytes, (byte) 'a');
+ System.arraycopy(prefix, 0, bytes, 0, prefix.length);
+ System.arraycopy(suffix, 0, bytes, OBJECT_SIZE - suffix.length, suffix.length);
+ return new ByteArrayValue(bytes);
+ }
+
+ private static final class ByteArrayValue implements IValueReference {
+ private final byte[] bytes;
+
+ private ByteArrayValue(byte[] bytes) {
+ this.bytes = bytes;
+ }
+
+ @Override
+ public byte[] getByteArray() {
+ return bytes;
+ }
+
+ @Override
+ public int getStartOffset() {
+ return 0;
+ }
+
+ @Override
+ public int getLength() {
+ return bytes.length;
+ }
+ }
+
+ /** Passes bytes straight to the stream, exactly like the textual printers. */
+ private static final class RawBytePrinter implements IExternalPrinter {
+ int sizeQueries;
+ private OutputStream outputStream;
+
+ @Override
+ public void open() {
+ }
+
+ @Override
+ public void newStream(OutputStream newOutputStream) throws HyracksDataException {
+ flushAndCloseCurrent();
+ outputStream = newOutputStream;
+ }
+
+ @Override
+ public long getBytesWritten() {
+ sizeQueries++;
+ return 0;
+ }
+
+ @Override
+ public void print(IValueReference value) throws HyracksDataException {
+ try {
+ outputStream.write(value.getByteArray(), value.getStartOffset(), value.getLength());
+ } catch (IOException e) {
+ throw HyracksDataException.create(e);
+ }
+ }
+
+ @Override
+ public void close() throws HyracksDataException {
+ flushAndCloseCurrent();
+ }
+
+ private void flushAndCloseCurrent() throws HyracksDataException {
+ if (outputStream == null) {
+ return;
+ }
+ try {
+ outputStream.flush();
+ outputStream.close();
+ } catch (IOException e) {
+ throw HyracksDataException.create(e);
+ }
+ outputStream = null;
+ }
+ }
+}
diff --git a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java
index 8121f6a..5ba4325 100644
--- a/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java
+++ b/asterixdb/asterix-metadata/src/main/java/org/apache/asterix/metadata/provider/ExternalWriterProvider.java
@@ -116,6 +116,14 @@
return Integer.parseInt(maxResultString);
}
+ private static long getMaxFileSize(IWriteDataSink sink) {
+ String maxFileSizeString = sink.getConfiguration().get(ExternalDataConstants.KEY_WRITER_MAX_FILE_SIZE);
+ if (maxFileSizeString == null) {
+ return ExternalDataConstants.WRITER_MAX_FILE_SIZE_UNLIMITED;
+ }
+ return StorageUtil.getByteValue(maxFileSizeString);
+ }
+
private static int getMaxParquetSchema(Map<String, String> conf) {
String maxResultString = conf.get(ExternalDataConstants.PARQUET_MAX_SCHEMAS_KEY);
if (maxResultString == null) {
@@ -159,6 +167,7 @@
fileWriterFactory.validate(appCtx);
String fileExtension = ExternalWriterProvider.getFileExtension(sink);
int maxResult = ExternalWriterProvider.getMaxResult(sink);
+ long maxFileSize = ExternalWriterProvider.getMaxFileSize(sink);
Map<String, String> configuration = sink.getConfiguration();
String format = configuration.get(ExternalDataConstants.KEY_FORMAT);
@@ -184,7 +193,7 @@
printerFactory = CleanJSONPrinterFactoryProvider.INSTANCE.getPrinterFactory(sourceType);
externalPrinterFactory = new TextualExternalFilePrinterFactory(printerFactory, compressStreamFactory);
writerFactory = new ExternalFileWriterFactory(fileWriterFactory, externalPrinterFactory,
- pathResolverFactory, maxResult);
+ pathResolverFactory, maxResult, maxFileSize);
return new SinkExternalWriterRuntimeFactory(sourceColumn, partitionColumns,
partitionComparatorFactories, inputDesc, writerFactory);
@@ -200,7 +209,7 @@
externalPrinterFactory =
new CsvExternalFilePrinterFactory(printerFactory, compressStreamFactory);
writerFactory = new ExternalFileWriterFactory(fileWriterFactory, externalPrinterFactory,
- pathResolverFactory, maxResult);
+ pathResolverFactory, maxResult, maxFileSize);
return new SinkExternalWriterRuntimeFactory(sourceColumn, partitionColumns,
partitionComparatorFactories, inputDesc, writerFactory);
} else {
@@ -234,7 +243,7 @@
(IAType) sourceType, rowGroupSize, pageSize, writerVersion);
ExternalFileWriterFactory parquetWriterFactory = new ExternalFileWriterFactory(fileWriterFactory,
- parquetPrinterFactory, pathResolverFactory, maxResult);
+ parquetPrinterFactory, pathResolverFactory, maxResult, maxFileSize);
return new SinkExternalWriterRuntimeFactory(sourceColumn, partitionColumns,
partitionComparatorFactories, inputDesc, parquetWriterFactory);
}
@@ -245,8 +254,8 @@
new ParquetExternalFilePrinterFactoryProvider(compressionCodecName, (IAType) sourceType,
rowGroupSize, pageSize, writerVersion);
return new ParquetSinkExternalWriterFactory(partitionerFactory, inputDesc, sourceColumn,
- (IAType) sourceType, maxSchemas, fileWriterFactory, maxResult, printerFactoryProvider,
- pathResolverFactory);
+ (IAType) sourceType, maxSchemas, fileWriterFactory, maxResult, maxFileSize,
+ printerFactoryProvider, pathResolverFactory);
default:
throw new CompilationException(ErrorCode.UNSUPPORTED_WRITING_FORMAT, pathSourceLocation, format,
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriter.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriter.java
index e8bae00..1ed0d31 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriter.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriter.java
@@ -28,13 +28,24 @@
private final IPathResolver pathResolver;
private final IExternalFileWriter writer;
private final int maxResultPerFile;
+ private final long maxFileSize;
+ private final boolean fileSizeBounded;
private String partitionPath;
private int tupleCounter;
- public ExternalFileWriter(IPathResolver pathResolver, IExternalFileWriter writer, int maxResultPerFile) {
+ /**
+ * @param pathResolver resolves partition directories and file names
+ * @param writer the file writer to delegate to
+ * @param maxResultPerFile maximum objects per file
+ * @param maxFileSize maximum size in bytes of a single file, or any non-positive value for no limit
+ */
+ public ExternalFileWriter(IPathResolver pathResolver, IExternalFileWriter writer, int maxResultPerFile,
+ long maxFileSize) {
this.pathResolver = pathResolver;
this.writer = writer;
this.maxResultPerFile = maxResultPerFile;
+ this.maxFileSize = maxFileSize;
+ this.fileSizeBounded = maxFileSize > 0;
}
@Override
@@ -62,7 +73,7 @@
// e.g., if max is 1000, we hit tuple 1001, we will upload and create a new file, if we only have 1000
// we will stop here, and calling the close/finish will upload whatever is written. This is to avoid
// creating and uploading empty files
- if (tupleCounter >= maxResultPerFile) {
+ if (tupleCounter >= maxResultPerFile || (fileSizeBounded && writer.getBytesWritten() >= maxFileSize)) {
newFile();
}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriterFactory.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriterFactory.java
index f85a8ab..5fd093d 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriterFactory.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/ExternalFileWriterFactory.java
@@ -28,20 +28,22 @@
private final IExternalFileWriterFactory writerFactory;
private final IExternalPrinterFactory printerFactory;
private final int maxResult;
+ private final long maxFileSize;
private final IPathResolverFactory pathResolverFactory;
public ExternalFileWriterFactory(IExternalFileWriterFactory writerFactory, IExternalPrinterFactory printerFactory,
- IPathResolverFactory pathResolverFactory, int maxResult) {
+ IPathResolverFactory pathResolverFactory, int maxResult, long maxFileSize) {
this.writerFactory = writerFactory;
this.printerFactory = printerFactory;
this.pathResolverFactory = pathResolverFactory;
this.maxResult = maxResult;
+ this.maxFileSize = maxFileSize;
}
@Override
public IExternalWriter createWriter(IHyracksTaskContext context) throws HyracksDataException {
IPathResolver resolver = pathResolverFactory.createResolver(context);
IExternalFileWriter writer = writerFactory.createWriter(context, printerFactory);
- return new ExternalFileWriter(resolver, writer, maxResult);
+ return new ExternalFileWriter(resolver, writer, maxResult, maxFileSize);
}
}
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalFileWriter.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalFileWriter.java
index f8ae81b..0f12014 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalFileWriter.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalFileWriter.java
@@ -55,6 +55,14 @@
void write(IValueReference value) throws HyracksDataException;
/**
+ * Size of the file currently being written. The value is approximate: it reflects the bytes accepted for
+ * the target after compression, and may exclude bytes that are only produced when the file is closed.
+ *
+ * @return bytes written to the current file, or 0 if no file is open
+ */
+ long getBytesWritten() throws HyracksDataException;
+
+ /**
* Run the abort sequence in case of a failure
*/
void abort() throws HyracksDataException;
diff --git a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalPrinter.java b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalPrinter.java
index 38ede08..4157f17 100644
--- a/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalPrinter.java
+++ b/asterixdb/asterix-runtime/src/main/java/org/apache/asterix/runtime/writer/IExternalPrinter.java
@@ -47,6 +47,15 @@
void newStream(OutputStream outputStream) throws HyracksDataException;
/**
+ * Size of the current file as accounted for by the printer, including anything it still buffers
+ *
+ * @return the printer's byte count, or 0 for printers that pass everything straight to the stream
+ */
+ default long getBytesWritten() {
+ return 0;
+ }
+
+ /**
* Print the provided value
*
* @param value to print
diff --git a/asterixdb/asterix-runtime/src/test/java/org/apache/asterix/runtime/writer/ExternalFileWriterMaxFileSizeTest.java b/asterixdb/asterix-runtime/src/test/java/org/apache/asterix/runtime/writer/ExternalFileWriterMaxFileSizeTest.java
new file mode 100644
index 0000000..2456293
--- /dev/null
+++ b/asterixdb/asterix-runtime/src/test/java/org/apache/asterix/runtime/writer/ExternalFileWriterMaxFileSizeTest.java
@@ -0,0 +1,185 @@
+/*
+ * 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.asterix.runtime.writer;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.data.std.api.IValueReference;
+import org.apache.hyracks.data.std.primitive.VoidPointable;
+import org.apache.hyracks.dataflow.common.data.accessors.IFrameTupleReference;
+import org.junit.Test;
+
+public class ExternalFileWriterMaxFileSizeTest {
+
+ private static final int NO_MAX_RESULT = Integer.MAX_VALUE;
+ private static final long NO_MAX_FILE_SIZE = -1L;
+ private static final String DIRECTORY = "dir/";
+
+ @Test
+ public void testSizeUnboundedByDefault() throws HyracksDataException {
+ CountingFileWriter fileWriter = new CountingFileWriter(1024);
+ write(fileWriter, NO_MAX_RESULT, NO_MAX_FILE_SIZE, 100);
+ assertEquals(1, fileWriter.files.size());
+ }
+
+ @Test
+ public void testRollsOverOnSizeLimit() throws HyracksDataException {
+ // 10 objects of 100 bytes each with a 200 byte limit: the limit divides evenly, so each file is
+ // closed exactly on it
+ CountingFileWriter fileWriter = new CountingFileWriter(100);
+ write(fileWriter, NO_MAX_RESULT, 200, 10);
+ assertEquals(List.of(2, 2, 2, 2, 2), fileWriter.files);
+ }
+
+ @Test
+ public void testFileOvershootsWhenObjectCrossesLimit() throws HyracksDataException {
+ // 100 byte objects against a 250 byte limit: the third object takes the file to 300 bytes, past the
+ // limit, because the limit can only be enforced between objects
+ CountingFileWriter fileWriter = new CountingFileWriter(100);
+ write(fileWriter, NO_MAX_RESULT, 250, 9);
+ assertEquals(List.of(3, 3, 3), fileWriter.files);
+ }
+
+ @Test
+ public void testCountLimitStillWinsWhenSmaller() throws HyracksDataException {
+ CountingFileWriter fileWriter = new CountingFileWriter(1);
+ write(fileWriter, 4, 1024, 12);
+ assertEquals(List.of(4, 4, 4), fileWriter.files);
+ }
+
+ @Test
+ public void testSizeLimitStillWinsWhenSmaller() throws HyracksDataException {
+ CountingFileWriter fileWriter = new CountingFileWriter(100);
+ write(fileWriter, 1000, 200, 9);
+ assertEquals(List.of(2, 2, 2, 2, 1), fileWriter.files);
+ }
+
+ @Test
+ public void testObjectLargerThanLimitIsWrittenAlone() throws HyracksDataException {
+ // the limit is consulted before writing, so an oversized object lands in a file of its own that
+ // overshoots the limit rather than being split or rejected
+ CountingFileWriter fileWriter = new CountingFileWriter(4096);
+ write(fileWriter, NO_MAX_RESULT, 1024, 3);
+ assertEquals(List.of(1, 1, 1), fileWriter.files);
+ }
+
+ @Test
+ public void testNoEmptyFileIsCreated() throws HyracksDataException {
+ CountingFileWriter fileWriter = new CountingFileWriter(100);
+ write(fileWriter, NO_MAX_RESULT, 200, 10);
+ for (int objectsInFile : fileWriter.files) {
+ assertTrue("a file was created without any object in it", objectsInFile > 0);
+ }
+ }
+
+ @Test
+ public void testSizeIsNotConsultedWhenUnbounded() throws HyracksDataException {
+ CountingFileWriter fileWriter = new CountingFileWriter(1024);
+ write(fileWriter, NO_MAX_RESULT, NO_MAX_FILE_SIZE, 100);
+ assertEquals(0, fileWriter.sizeQueries);
+ }
+
+ @Test
+ public void testSizeIsConsultedForEveryObject() throws HyracksDataException {
+ // the limit is enforced as tightly as it can be: the size is read once per object, so the file can
+ // only ever overshoot by the object that crosses the limit
+ CountingFileWriter fileWriter = new CountingFileWriter(1);
+ int objects = 10000;
+ write(fileWriter, NO_MAX_RESULT, 1024 * 1024, objects);
+ assertEquals(1, fileWriter.files.size());
+ assertEquals(objects, fileWriter.sizeQueries);
+ }
+
+ private static void write(CountingFileWriter fileWriter, int maxResultPerFile, long maxFileSize, int objects)
+ throws HyracksDataException {
+ ExternalFileWriter writer =
+ new ExternalFileWriter(new FixedPathResolver(), fileWriter, maxResultPerFile, maxFileSize);
+ writer.open();
+ writer.initNewPartition(null);
+ for (int i = 0; i < objects; i++) {
+ writer.write(new VoidPointable());
+ }
+ writer.close();
+ }
+
+ private static class FixedPathResolver implements IPathResolver {
+ private int fileCounter;
+
+ @Override
+ public String getPartitionDirectory(IFrameTupleReference tuple) {
+ return DIRECTORY;
+ }
+
+ @Override
+ public String getNextFileName() {
+ return "file-" + fileCounter++;
+ }
+ }
+
+ /**
+ * Reports a fixed number of bytes per written object and records how many objects landed in each file.
+ */
+ private static class CountingFileWriter implements IExternalFileWriter {
+ private final List<Integer> files = new ArrayList<>();
+ private final int bytesPerObject;
+ private int sizeQueries;
+
+ CountingFileWriter(int bytesPerObject) {
+ this.bytesPerObject = bytesPerObject;
+ }
+
+ @Override
+ public void open() {
+ }
+
+ @Override
+ public void validate(String directory) {
+ }
+
+ @Override
+ public boolean newFile(String directory, String fileName) {
+ files.add(0);
+ return true;
+ }
+
+ @Override
+ public void write(IValueReference value) {
+ files.set(files.size() - 1, files.get(files.size() - 1) + 1);
+ }
+
+ @Override
+ public long getBytesWritten() {
+ sizeQueries++;
+ return (long) files.get(files.size() - 1) * bytesPerObject;
+ }
+
+ @Override
+ public void abort() {
+ }
+
+ @Override
+ public void close() {
+ }
+ }
+}
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21471?usp=email
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: I44e6777175c507b1c289fd7cb8985e23331fdd49
Gerrit-Change-Number: 21471
Gerrit-PatchSet: 1
Gerrit-Owner: Rithwik Koul <ri...@couchbase.com>