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 2025/07/08 13:31:00 UTC
Change in asterixdb[master]: [NO ISSUE][RT]: Refactor: Remove code duplication in stable sort oper...
From Janhavi Tripurwar <ja...@couchbase.com>:
Janhavi Tripurwar has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/20035 )
Change subject: [NO ISSUE][RT]: Refactor: Remove code duplication in stable sort operators
......................................................................
[NO ISSUE][RT]: Refactor: Remove code duplication in stable sort operators
Change-Id: I2a0fc093d356be7e2379a66c6843ebd6b6cdbd4c
---
M hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroStableSortPOperator.java
M hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/StableSortPOperator.java
A hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SortOperatorHelper.java
3 files changed, 108 insertions(+), 74 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/35/20035/1
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroStableSortPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroStableSortPOperator.java
index 403ca57..d974c88 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroStableSortPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/MicroStableSortPOperator.java
@@ -21,21 +21,11 @@
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.algebricks.core.algebra.base.IHyracksJobBuilder;
import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
-import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
import org.apache.hyracks.algebricks.core.algebra.base.PhysicalOperatorTag;
-import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.IOperatorSchema;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.OrderOperator.IOrder.OrderKind;
-import org.apache.hyracks.algebricks.core.algebra.properties.OrderColumn;
import org.apache.hyracks.algebricks.core.jobgen.impl.JobGenContext;
-import org.apache.hyracks.algebricks.core.jobgen.impl.JobGenHelper;
-import org.apache.hyracks.algebricks.data.IBinaryComparatorFactoryProvider;
-import org.apache.hyracks.algebricks.data.INormalizedKeyComputerFactoryProvider;
import org.apache.hyracks.algebricks.runtime.base.IPushRuntimeFactory;
import org.apache.hyracks.algebricks.runtime.operators.sort.MicroSortRuntimeFactory;
-import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
-import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputerFactory;
-import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
public class MicroStableSortPOperator extends AbstractStableSortPOperator {
@@ -56,32 +46,12 @@
public void contributeRuntimeOperator(IHyracksJobBuilder builder, JobGenContext context, ILogicalOperator op,
IOperatorSchema opSchema, IOperatorSchema[] inputSchemas, IOperatorSchema outerPlanSchema)
throws AlgebricksException {
- RecordDescriptor recDescriptor =
- JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op), opSchema, context);
- int n = sortColumns.length;
- int[] sortFields = new int[n];
- IBinaryComparatorFactory[] comps = new IBinaryComparatorFactory[n];
- int i = 0;
- INormalizedKeyComputerFactoryProvider nkcfProvider = context.getNormalizedKeyComputerFactoryProvider();
- INormalizedKeyComputerFactory nkcf = null;
- IVariableTypeEnvironment env = context.getTypeEnvironment(op);
- for (OrderColumn oc : sortColumns) {
- LogicalVariable var = oc.getColumn();
- sortFields[i] = opSchema.findVariable(var);
- Object type = env.getVarType(var);
- OrderKind order = oc.getOrder();
- if (i == 0 && nkcfProvider != null && type != null) {
- nkcf = nkcfProvider.getNormalizedKeyComputerFactory(type, order == OrderKind.ASC);
- }
+ SortOperatorHelper.SortSetupData sortSetupData =
+ SortOperatorHelper.setupSortOperator(context, op, opSchema, sortColumns, localMemoryRequirements);
- IBinaryComparatorFactoryProvider bcfp = context.getBinaryComparatorFactoryProvider();
- comps[i] = bcfp.getBinaryComparatorFactory(type, oc.getOrder() == OrderKind.ASC);
- i++;
- }
-
- int maxNumberOfFrames = localMemoryRequirements.getMemoryBudgetInFrames();
- IPushRuntimeFactory runtime = new MicroSortRuntimeFactory(sortFields, nkcf, comps, null, maxNumberOfFrames);
- builder.contributeMicroOperator(op, runtime, recDescriptor);
+ IPushRuntimeFactory runtime = new MicroSortRuntimeFactory(sortSetupData.sortFields, sortSetupData.nkcf,
+ sortSetupData.comps, null, sortSetupData.maxNumberOfFrames);
+ builder.contributeMicroOperator(op, runtime, sortSetupData.recDescriptor);
ILogicalOperator src = op.getInputs().get(0).getValue();
builder.contributeGraphEdge(src, 0, op, 0);
}
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SortOperatorHelper.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SortOperatorHelper.java
new file mode 100644
index 0000000..b6487fe
--- /dev/null
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/SortOperatorHelper.java
@@ -0,0 +1,88 @@
+/*
+ * 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.algebricks.core.algebra.operators.physical;
+
+import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
+import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
+import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
+import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.IOperatorSchema;
+import org.apache.hyracks.algebricks.core.algebra.operators.logical.OrderOperator;
+import org.apache.hyracks.algebricks.core.algebra.properties.LocalMemoryRequirements;
+import org.apache.hyracks.algebricks.core.algebra.properties.OrderColumn;
+import org.apache.hyracks.algebricks.core.jobgen.impl.JobGenContext;
+import org.apache.hyracks.algebricks.core.jobgen.impl.JobGenHelper;
+import org.apache.hyracks.algebricks.data.IBinaryComparatorFactoryProvider;
+import org.apache.hyracks.algebricks.data.INormalizedKeyComputerFactoryProvider;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputerFactory;
+import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
+
+public class SortOperatorHelper {
+
+ public static class SortSetupData {
+ public final int[] sortFields;
+ public final IBinaryComparatorFactory[] comps;
+ public final INormalizedKeyComputerFactory nkcf;
+ public final RecordDescriptor recDescriptor;
+ public final int maxNumberOfFrames;
+
+ public SortSetupData(int[] sortFields, IBinaryComparatorFactory[] comps, INormalizedKeyComputerFactory nkcf,
+ RecordDescriptor recDescriptor, int maxNumberOfFrames) {
+ this.sortFields = sortFields;
+ this.comps = comps;
+ this.nkcf = nkcf;
+ this.recDescriptor = recDescriptor;
+ this.maxNumberOfFrames = maxNumberOfFrames;
+ }
+ }
+
+ public static SortSetupData setupSortOperator(JobGenContext context, ILogicalOperator op, IOperatorSchema opSchema,
+ OrderColumn[] sortColumns, LocalMemoryRequirements memReq) throws AlgebricksException {
+
+ RecordDescriptor recDescriptor =
+ JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op), opSchema, context);
+ int n = sortColumns.length;
+ int[] sortFields = new int[n];
+ IBinaryComparatorFactory[] comps = new IBinaryComparatorFactory[n];
+
+ INormalizedKeyComputerFactoryProvider nkcfProvider = context.getNormalizedKeyComputerFactoryProvider();
+ INormalizedKeyComputerFactory nkcf = null;
+ IVariableTypeEnvironment env = context.getTypeEnvironment(op);
+
+ for (int i = 0; i < n; i++) {
+ OrderColumn oc = sortColumns[i];
+ LogicalVariable var = oc.getColumn();
+ sortFields[i] = opSchema.findVariable(var);
+ Object type = env.getVarType(var);
+ OrderOperator.IOrder.OrderKind order = oc.getOrder();
+
+ if (i == 0 && nkcfProvider != null && type != null) {
+ nkcf = nkcfProvider.getNormalizedKeyComputerFactory(type, order == OrderOperator.IOrder.OrderKind.ASC);
+ }
+
+ IBinaryComparatorFactoryProvider bcfp = context.getBinaryComparatorFactoryProvider();
+ comps[i] = bcfp.getBinaryComparatorFactory(type, oc.getOrder() == OrderOperator.IOrder.OrderKind.ASC);
+ }
+
+ int maxNumberOfFrames = memReq.getMemoryBudgetInFrames();
+ return new SortSetupData(sortFields, comps, nkcf, recDescriptor, maxNumberOfFrames);
+ }
+}
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/StableSortPOperator.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/StableSortPOperator.java
index 93c5c3b..416ed9e 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/StableSortPOperator.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/operators/physical/StableSortPOperator.java
@@ -21,21 +21,11 @@
import org.apache.hyracks.algebricks.common.exceptions.AlgebricksException;
import org.apache.hyracks.algebricks.core.algebra.base.IHyracksJobBuilder;
import org.apache.hyracks.algebricks.core.algebra.base.ILogicalOperator;
-import org.apache.hyracks.algebricks.core.algebra.base.LogicalVariable;
import org.apache.hyracks.algebricks.core.algebra.base.PhysicalOperatorTag;
-import org.apache.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.IOperatorSchema;
import org.apache.hyracks.algebricks.core.algebra.operators.logical.OrderOperator;
-import org.apache.hyracks.algebricks.core.algebra.operators.logical.OrderOperator.IOrder.OrderKind;
-import org.apache.hyracks.algebricks.core.algebra.properties.OrderColumn;
import org.apache.hyracks.algebricks.core.jobgen.impl.JobGenContext;
-import org.apache.hyracks.algebricks.core.jobgen.impl.JobGenHelper;
-import org.apache.hyracks.algebricks.data.IBinaryComparatorFactoryProvider;
-import org.apache.hyracks.algebricks.data.INormalizedKeyComputerFactoryProvider;
-import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
-import org.apache.hyracks.api.dataflow.value.INormalizedKeyComputerFactory;
-import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
import org.apache.hyracks.dataflow.std.sort.AbstractSorterOperatorDescriptor;
import org.apache.hyracks.dataflow.std.sort.ExternalSortOperatorDescriptor;
@@ -72,42 +62,19 @@
IOperatorSchema opSchema, IOperatorSchema[] inputSchemas, IOperatorSchema outerPlanSchema)
throws AlgebricksException {
IOperatorDescriptorRegistry spec = builder.getJobSpec();
- RecordDescriptor recDescriptor =
- JobGenHelper.mkRecordDescriptor(context.getTypeEnvironment(op), opSchema, context);
- int n = sortColumns.length;
- int[] sortFields = new int[n];
- IBinaryComparatorFactory[] comps = new IBinaryComparatorFactory[n];
+ SortOperatorHelper.SortSetupData sortSetupData =
+ SortOperatorHelper.setupSortOperator(context, op, opSchema, sortColumns, localMemoryRequirements);
- INormalizedKeyComputerFactoryProvider nkcfProvider = context.getNormalizedKeyComputerFactoryProvider();
- INormalizedKeyComputerFactory nkcf = null;
-
- IVariableTypeEnvironment env = context.getTypeEnvironment(op);
- int i = 0;
- // TODO(ali): should refactor common code with micro sort op
- for (OrderColumn oc : sortColumns) {
- LogicalVariable var = oc.getColumn();
- sortFields[i] = opSchema.findVariable(var);
- Object type = env.getVarType(var);
- OrderKind order = oc.getOrder();
- if (i == 0 && nkcfProvider != null && type != null) {
- nkcf = nkcfProvider.getNormalizedKeyComputerFactory(type, order == OrderKind.ASC);
- }
- IBinaryComparatorFactoryProvider bcfp = context.getBinaryComparatorFactoryProvider();
- comps[i] = bcfp.getBinaryComparatorFactory(type, oc.getOrder() == OrderKind.ASC);
- i++;
- }
-
- int maxNumberOfFrames = localMemoryRequirements.getMemoryBudgetInFrames();
AbstractSorterOperatorDescriptor sortOpDesc;
// topK == -1 means that a topK value is not provided.
if (topK == -1) {
- sortOpDesc =
- new ExternalSortOperatorDescriptor(spec, maxNumberOfFrames, sortFields, nkcf, comps, recDescriptor);
+ sortOpDesc = new ExternalSortOperatorDescriptor(spec, sortSetupData.maxNumberOfFrames,
+ sortSetupData.sortFields, sortSetupData.nkcf, sortSetupData.comps, sortSetupData.recDescriptor);
} else {
// Since topK value is provided, topK optimization is possible.
// We call topKSorter instead of calling ExternalSortOperator.
- sortOpDesc = new TopKSorterOperatorDescriptor(spec, maxNumberOfFrames, topK, sortFields, nkcf, comps,
- recDescriptor);
+ sortOpDesc = new TopKSorterOperatorDescriptor(spec, sortSetupData.maxNumberOfFrames, topK,
+ sortSetupData.sortFields, sortSetupData.nkcf, sortSetupData.comps, sortSetupData.recDescriptor);
}
sortOpDesc.setSourceLocation(op.getSourceLocation());
contributeOpDesc(builder, (AbstractLogicalOperator) op, sortOpDesc);
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/20035
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: I2a0fc093d356be7e2379a66c6843ebd6b6cdbd4c
Gerrit-Change-Number: 20035
Gerrit-PatchSet: 1
Gerrit-Owner: Janhavi Tripurwar <ja...@couchbase.com>
Gerrit-MessageType: newchange