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 2022/07/29 01:36:19 UTC

Change in asterixdb[master]: Join Intermediate Reads and Writes

From Shiva Jahangiri <sh...@uci.edu>:

Shiva Jahangiri has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17023 )


Change subject: Join Intermediate Reads and Writes
......................................................................

Join Intermediate Reads and Writes

Change-Id: I1fd519fac1cd62a1851714de4fdccca949b53971
---
A hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/JoinStats.java
M hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/OptimizedHybridHashJoin.java
M hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/OptimizedHybridHashJoinOperatorDescriptor.java
3 files changed, 93 insertions(+), 4 deletions(-)



  git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/23/17023/1

diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/JoinStats.java b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/JoinStats.java
new file mode 100644
index 0000000..f4758b6
--- /dev/null
+++ b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/JoinStats.java
@@ -0,0 +1,54 @@
+/*
+ * 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.hyracks.dataflow.std.join;
+
+import java.io.Serializable;
+
+public class JoinStats implements Serializable {
+    private long writes;
+    private long reads;
+
+    public long getWrites() {
+        return writes;
+    }
+
+    public long getReads() {
+        return reads;
+    }
+
+    public void addToWrites(long write) {
+        this.writes += write;
+    }
+
+    public void addToReads(long read) {
+        this.reads += read;
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder stb = new StringBuilder();
+        stb.append("Writes: " + this.writes + " (bytes) " + "Reads: " + (this.reads + " (bytes) "));
+        return stb.toString();
+    }
+
+    public void addAll(JoinStats other) {
+        this.writes += other.getWrites();
+        this.reads += other.getReads();
+    }
+}
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/OptimizedHybridHashJoin.java b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/OptimizedHybridHashJoin.java
index 7a9bb25..9849be8 100644
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/OptimizedHybridHashJoin.java
+++ b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/OptimizedHybridHashJoin.java
@@ -91,6 +91,8 @@
     // corresponding function signature.
     private final TuplePointer tempPtr = new TuplePointer();
     private int[] probePSizeInTups;
+    private JoinStats buildStats;
+    private JoinStats probeStats;
 
     public OptimizedHybridHashJoin(IHyracksJobletContext jobletCtx, int memSizeInFrames, int numOfPartitions,
             String probeRelName, String buildRelName, RecordDescriptor probeRd, RecordDescriptor buildRd,
@@ -117,6 +119,8 @@
         this.probePredEval = probePredEval;
         this.isReversed = false;
         this.spilledStatus = new BitSet(numOfPartitions);
+        this.buildStats = new JoinStats();
+        this.probeStats = new JoinStats();
         this.nonMatchWriters = isLeftOuter ? new IMissingWriter[nullWriterFactories1.length] : null;
         if (isLeftOuter) {
             for (int i = 0; i < nullWriterFactories1.length; i++) {
@@ -192,6 +196,7 @@
         if (buildRFWriters[pid] == null) {
             throw new HyracksDataException("Tried to close the non-existing file writer.");
         }
+        buildStats.addToWrites(buildRFWriters[pid].getFileSize());
         buildRFWriters[pid].close();
     }
 
@@ -209,7 +214,7 @@
 
     public void closeBuild() throws HyracksDataException {
         // Flushes the remaining chunks of the all spilled partitions to the disk.
-        closeAllSpilledPartitions(buildRFWriters, buildRelName);
+        closeAllSpilledPartitions(buildRFWriters, buildRelName, buildStats);
 
         // Makes the space for the in-memory hash table (some partitions may need to be spilled to the disk
         // during this step in order to make the space.)
@@ -256,7 +261,8 @@
         }
     }
 
-    private void closeAllSpilledPartitions(RunFileWriter[] runFileWriters, String refName) throws HyracksDataException {
+    private void closeAllSpilledPartitions(RunFileWriter[] runFileWriters, String refName, JoinStats stats)
+            throws HyracksDataException {
         try {
             for (int pid = spilledStatus.nextSetBit(0); pid >= 0 && pid < numOfPartitions; pid =
                     spilledStatus.nextSetBit(pid + 1)) {
@@ -270,6 +276,7 @@
             // Force to close all run file writers.
             for (RunFileWriter runFileWriter : runFileWriters) {
                 if (runFileWriter != null) {
+                    stats.addToWrites(runFileWriter.getFileSize());
                     runFileWriter.close();
                 }
             }
@@ -339,6 +346,7 @@
         long currentFreeSpace = freeSpace;
         while ((pid = selectAPartitionToReload(currentFreeSpace, pid, currentMemoryTupleCount)) >= 0
                 && loadSpilledPartitionToMem(pid, buildRFWriters[pid])) {
+            buildStats.addToReads(buildRFWriters[pid].getFileSize());
             currentMemoryTupleCount += buildPSizeInTups[pid];
             // Reserve space for loaded data & increase in hash table (give back one frame taken by spilled partition.)
             currentFreeSpace = currentFreeSpace
@@ -558,7 +566,7 @@
 
     public void releaseResource() throws HyracksDataException {
         inMemJoiner.closeTable();
-        closeAllSpilledPartitions(probeRFWriters, probeRelName);
+        closeAllSpilledPartitions(probeRFWriters, probeRelName, probeStats);
         bufferManager.close();
         inMemJoiner = null;
         bufferManager = null;
@@ -613,4 +621,12 @@
         }
         this.isReversed = reversed;
     }
+
+    public JoinStats getBuildStats() {
+        return buildStats;
+    }
+
+    public JoinStats getProbeStats() {
+        return probeStats;
+    }
 }
diff --git a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/OptimizedHybridHashJoinOperatorDescriptor.java b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/OptimizedHybridHashJoinOperatorDescriptor.java
index 555e8fb..87411eb 100644
--- a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/OptimizedHybridHashJoinOperatorDescriptor.java
+++ b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/join/OptimizedHybridHashJoinOperatorDescriptor.java
@@ -22,6 +22,7 @@
 import java.io.DataOutput;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.BitSet;
 
 import org.apache.hyracks.api.comm.IFrame;
@@ -112,7 +113,8 @@
      * See https://issues.apache.org/jira/browse/ASTERIXDB-2783 for more details.
      */
     private static final int INIT_SEED = 982028031;
-
+    private JoinStats buildStats;
+    private JoinStats probeStats;
     private static final int BUILD_AND_PARTITION_ACTIVITY_ID = 0;
     private static final int PARTITION_AND_JOIN_ACTIVITY_ID = 1;
 
@@ -167,6 +169,8 @@
         this.buildPredEvalFactory = predEvalFactory1;
         this.isLeftOuter = isLeftOuter;
         this.nonMatchWriterFactories = nonMatchWriterFactories;
+        this.buildStats = new JoinStats();
+        this.probeStats = new JoinStats();
     }
 
     public OptimizedHybridHashJoinOperatorDescriptor(IOperatorDescriptorRegistry spec, int memSizeInFrames,
@@ -229,6 +233,7 @@
         private int memForJoin;
         private int numOfPartitions;
         private OptimizedHybridHashJoin hybridHJ;
+        private ArrayList<OptimizedHybridHashJoin> partitionJoins = new ArrayList<>();
 
         public BuildAndPartitionTaskState() {
         }
@@ -464,6 +469,16 @@
                             int pSize = state.hybridHJ.getProbePartitionSizeInTup(pid);
                             joinPartitionPair(bReader, pReader, bSize, pSize, 1);
                         }
+                        buildStats.addAll(state.hybridHJ.getBuildStats());
+                        probeStats.addAll(state.hybridHJ.getProbeStats());
+                        for (OptimizedHybridHashJoin hhj : state.partitionJoins) {
+                            probeStats.addAll(hhj.getBuildStats());
+                            probeStats.addAll(hhj.getProbeStats());
+                        }
+                        LOGGER.info("BuildStats: " + buildStats.toString());
+                        LOGGER.info("ProbeStats: " + probeStats.toString());
+                        buildStats.addAll(probeStats);
+                        LOGGER.info("TotalStats: " + buildStats.toString());
                     } catch (Exception e) {
                         if (state.hybridHJ != null) {
                             state.hybridHJ.fail();
@@ -500,6 +515,8 @@
                     ITuplePartitionComputer buildHpc =
                             new FieldHashPartitionComputerFamily(buildKeys, buildHashFunctionFactories)
                                     .createPartitioner(level);
+                    probeStats.addToReads(buildSideReader.getFileSize());
+                    probeStats.addToReads(probeSideReader.getFileSize());
 
                     int frameSize = jobletCtx.getInitialFrameSize();
                     long buildPartSize = (long) Math.ceil((double) buildSideReader.getFileSize() / (double) frameSize);
@@ -599,6 +616,8 @@
                     rHHj = new OptimizedHybridHashJoin(jobletCtx, state.memForJoin, n, PROBE_REL, BUILD_REL, probeRd,
                             buildRd, probeHpc, buildHpc, null, null, isLeftOuter, nonMatchWriterFactories); //checked-confirmed
 
+                    state.partitionJoins.add(rHHj);
+
                     rHHj.setIsReversed(isReversed);
                     try {
                         buildSideReader.open();

-- 
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/17023
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings

Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: I1fd519fac1cd62a1851714de4fdccca949b53971
Gerrit-Change-Number: 17023
Gerrit-PatchSet: 1
Gerrit-Owner: Shiva Jahangiri <sh...@uci.edu>
Gerrit-MessageType: newchange