[
https://issues.apache.org/jira/browse/HIVE-25772?focusedWorklogId=720258&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-720258
]
ASF GitHub Bot logged work on HIVE-25772:
-----------------------------------------
Author: ASF GitHub Bot
Created on: 03/Feb/22 16:25
Start Date: 03/Feb/22 16:25
Worklog Time Spent: 10m
Work Description: marton-bod commented on a change in pull request #2842:
URL: https://github.com/apache/hive/pull/2842#discussion_r798741568
##########
File path:
iceberg/patched-iceberg-core/src/main/java/org/apache/iceberg/io/ClusteredWriter.java
##########
@@ -0,0 +1,159 @@
+/*
+ * 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.iceberg.io;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Comparator;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.encryption.EncryptedOutputFile;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Types.StructType;
+import org.apache.iceberg.util.StructLikeSet;
+
+/**
+ * Copied over from Iceberg 0.13.0 and has a difference of allowing bucket
transform partitions to arrive without
+ * any ordering. This is a temporary convenience solution until a permanent
one probably using UDFs is done.
+ *
+ * Original description:
+ *
+ * A writer capable of writing to multiple specs and partitions that requires
the incoming records
+ * to be clustered by partition spec and by partition within each spec.
+ * <p>
+ * As opposed to {@link FanoutWriter}, this writer keeps at most one file open
to reduce
+ * the memory consumption. Prefer using this writer whenever the incoming
records can be clustered
+ * by spec/partition.
+ */
+abstract class ClusteredWriter<T, R> implements PartitioningWriter<T, R> {
+
+ private static final Class<?> BUCKET_TRANSFORM_CLAZZ;
+
+ static {
+ try {
+ BUCKET_TRANSFORM_CLAZZ =
Class.forName("org.apache.iceberg.transforms.Bucket");
+ } catch (ClassNotFoundException e) {
+ throw new RuntimeException("Couldn't find Bucket transform class", e);
+ }
+ }
+
+ private static final String NOT_CLUSTERED_ROWS_ERROR_MSG_TEMPLATE =
+ "Incoming records violate the writer assumption that records are
clustered by spec and " +
+ "by partition within each spec. Either cluster the incoming records
or switch to fanout writers.\n" +
+ "Encountered records that belong to already closed files:\n";
+
+ private final Set<Integer> completedSpecIds = Sets.newHashSet();
+
+ private PartitionSpec currentSpec = null;
+ private Comparator<StructLike> partitionComparator = null;
+ private Set<StructLike> completedPartitions = null;
+ private StructLike currentPartition = null;
+ private FileWriter<T, R> currentWriter = null;
+
+ private boolean closed = false;
+
+ protected abstract FileWriter<T, R> newWriter(PartitionSpec spec, StructLike
partition);
+
+ protected abstract void addResult(R result);
+
+ protected abstract R aggregatedResult();
+
+ @Override
+ public void write(T row, PartitionSpec spec, StructLike partition) {
+ if (!spec.equals(currentSpec)) {
+ if (currentSpec != null) {
+ closeCurrentWriter();
+ completedSpecIds.add(currentSpec.specId());
+ completedPartitions.clear();
+ }
+
+ if (completedSpecIds.contains(spec.specId())) {
+ String errorCtx = String.format("spec %s", spec);
+ throw new IllegalStateException(NOT_CLUSTERED_ROWS_ERROR_MSG_TEMPLATE
+ errorCtx);
+ }
+
+ StructType partitionType = spec.partitionType();
+
+ this.currentSpec = spec;
+ this.partitionComparator = Comparators.forType(partitionType);
+ this.completedPartitions = StructLikeSet.create(partitionType);
+ // copy the partition key as the key object may be reused
+ this.currentPartition = StructCopy.copy(partition);
+ this.currentWriter = newWriter(currentSpec, currentPartition);
+
+ } else if (partition != currentPartition &&
partitionComparator.compare(partition, currentPartition) != 0) {
+ closeCurrentWriter();
+ completedPartitions.add(currentPartition);
+
+ if (completedPartitions.contains(partition) &&
!hasBucketTransform(currentSpec)) {
Review comment:
`!hasBucketTransform(currentSpec)`: this means that if the table has a
spec like `(a identity, date(b), bucket(16, c))`, then even if the out-of-order
record comes for `a` or `b`, we will still act leniently?
--
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: gitbox-unsubscr...@hive.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
-------------------
Worklog Id: (was: 720258)
Time Spent: 1h 10m (was: 1h)
> Use ClusteredWriter when writing to Iceberg tables
> --------------------------------------------------
>
> Key: HIVE-25772
> URL: https://issues.apache.org/jira/browse/HIVE-25772
> Project: Hive
> Issue Type: Improvement
> Reporter: Ádám Szita
> Assignee: Ádám Szita
> Priority: Major
> Labels: pull-request-available
> Time Spent: 1h 10m
> Remaining Estimate: 0h
>
> Currently Hive relies on PartitionedFanoutWriter to write records to Iceberg
> tables. This has a big disadvantage when it comes to writing many-many
> partitions, as it keeps a file handle open to each of the partitions.
> For some file systems like S3, there can be a big resource waste due to
> keeping bytebuffers for each of the handles, that may result in OOM scenarios.
> ClusteredWriter will only write one file at a time, and will not keep other
> file handles open, but it will expect that the input data is sorted by
> partition keys.
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
