Baunsgaard commented on code in PR #2466: URL: https://github.com/apache/systemds/pull/2466#discussion_r3249480935
########## src/main/java/org/apache/sysds/hops/estim/EstimatorRowWise.java: ########## @@ -0,0 +1,325 @@ +/* + * 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.sysds.hops.estim; + +import org.apache.commons.lang3.ArrayUtils; +import org.apache.commons.lang3.NotImplementedException; +import org.apache.sysds.hops.OptimizerUtils; +import org.apache.sysds.runtime.data.SparseRow; +import org.apache.sysds.runtime.matrix.data.MatrixBlock; +import org.apache.sysds.runtime.meta.DataCharacteristics; +import org.apache.sysds.runtime.meta.MatrixCharacteristics; + +import java.util.stream.DoubleStream; +import java.util.stream.IntStream; + +/** + * This estimator implements an approach based on row-wise sparsity estimation, + * introduced in + * Lin, Chunxu, Wensheng Luo, Yixiang Fang, Chenhao Ma, Xilin Liu and Yuchi Ma: + * On Efficient Large Sparse Matrix Chain Multiplication. + * Proceedings of the ACM on Management of Data 2 (2024): 1 - 27. + */ +public class EstimatorRowWise extends SparsityEstimator { + @Override + public DataCharacteristics estim(MMNode root) { + estimInternChain(root); + double sparsity = DoubleStream.of((double[])root.getSynopsis()).average().orElse(0); + + DataCharacteristics outputCharacteristics = deriveOutputCharacteristics(root, sparsity); + return root.setDataCharacteristics(outputCharacteristics); + } + + @Override + public double estim(MatrixBlock m1, MatrixBlock m2) { + return estim(m1, m2, OpCode.MM); + } + + @Override + public double estim(MatrixBlock m1, MatrixBlock m2, OpCode op) { + if( isExactMetadataOp(op, m1.getNumColumns()) ) { + return estimExactMetaData(m1.getDataCharacteristics(), + m2.getDataCharacteristics(), op).getSparsity(); + } + + double[] rsOut = estimIntern(m1, m2, op); + return DoubleStream.of(rsOut).average().orElse(0); + } + + @Override + public double estim(MatrixBlock m1, OpCode op) { + if( isExactMetadataOp(op, m1.getNumColumns()) ) + return estimExactMetaData(m1.getDataCharacteristics(), null, op).getSparsity(); + + double[] rsOut = estimIntern(m1, op); + return DoubleStream.of(rsOut).average().orElse(0); + } + + private void estimInternChain(MMNode node) { + estimInternChain(node, null, null); + } + + private void estimInternChain(MMNode node, double[] rsRightNeighbor, OpCode opRightNeighbor) { + double[] rsOut; + if(node.isLeaf()) { + MatrixBlock mb = node.getData(); + if(rsRightNeighbor != null) + rsOut = estimIntern(mb, rsRightNeighbor, opRightNeighbor); + else + rsOut = getRowWiseSparsityVector(mb); + } + else { + switch(node.getOp()) { + case MM: + estimInternChain(node.getRight(), rsRightNeighbor, opRightNeighbor); + estimInternChain(node.getLeft(), (double[])(node.getRight().getSynopsis()), node.getOp()); + rsOut = (double[])node.getLeft().getSynopsis(); + break; + case CBIND: + /** NOTE: considering the current node as new DAG for estimation (cut), since the row sparsity of + * the right neighbor cannot be aggregated into a cbind operation when having only row sparsity vectors + */ + estimInternChain(node.getLeft()); + estimInternChain(node.getRight()); + double[] rsCBind = estimInternCBind((double[])(node.getLeft().getSynopsis()), (double[])(node.getRight().getSynopsis())); + if(rsRightNeighbor != null) { + rsOut = (double[])estimInternMMFallback(rsCBind, rsRightNeighbor); + if(opRightNeighbor != OpCode.MM) + throw new NotImplementedException("Fallback sparsity estimation has only been " + + "considered for MM operation w/ right neighbor yet."); + } + else + rsOut = (double[])rsCBind; Review Comment: no need to cast to double[] ########## src/main/java/org/apache/sysds/hops/estim/EstimatorRowWise.java: ########## @@ -0,0 +1,325 @@ +/* + * 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.sysds.hops.estim; + +import org.apache.commons.lang3.ArrayUtils; +import org.apache.commons.lang3.NotImplementedException; +import org.apache.sysds.hops.OptimizerUtils; +import org.apache.sysds.runtime.data.SparseRow; +import org.apache.sysds.runtime.matrix.data.MatrixBlock; +import org.apache.sysds.runtime.meta.DataCharacteristics; +import org.apache.sysds.runtime.meta.MatrixCharacteristics; + +import java.util.stream.DoubleStream; +import java.util.stream.IntStream; + +/** + * This estimator implements an approach based on row-wise sparsity estimation, + * introduced in + * Lin, Chunxu, Wensheng Luo, Yixiang Fang, Chenhao Ma, Xilin Liu and Yuchi Ma: + * On Efficient Large Sparse Matrix Chain Multiplication. + * Proceedings of the ACM on Management of Data 2 (2024): 1 - 27. + */ +public class EstimatorRowWise extends SparsityEstimator { + @Override + public DataCharacteristics estim(MMNode root) { + estimInternChain(root); + double sparsity = DoubleStream.of((double[])root.getSynopsis()).average().orElse(0); + + DataCharacteristics outputCharacteristics = deriveOutputCharacteristics(root, sparsity); + return root.setDataCharacteristics(outputCharacteristics); + } + + @Override + public double estim(MatrixBlock m1, MatrixBlock m2) { + return estim(m1, m2, OpCode.MM); + } + + @Override + public double estim(MatrixBlock m1, MatrixBlock m2, OpCode op) { + if( isExactMetadataOp(op, m1.getNumColumns()) ) { + return estimExactMetaData(m1.getDataCharacteristics(), + m2.getDataCharacteristics(), op).getSparsity(); + } + + double[] rsOut = estimIntern(m1, m2, op); + return DoubleStream.of(rsOut).average().orElse(0); + } + + @Override + public double estim(MatrixBlock m1, OpCode op) { + if( isExactMetadataOp(op, m1.getNumColumns()) ) + return estimExactMetaData(m1.getDataCharacteristics(), null, op).getSparsity(); + + double[] rsOut = estimIntern(m1, op); + return DoubleStream.of(rsOut).average().orElse(0); + } + + private void estimInternChain(MMNode node) { + estimInternChain(node, null, null); + } + + private void estimInternChain(MMNode node, double[] rsRightNeighbor, OpCode opRightNeighbor) { + double[] rsOut; + if(node.isLeaf()) { + MatrixBlock mb = node.getData(); + if(rsRightNeighbor != null) + rsOut = estimIntern(mb, rsRightNeighbor, opRightNeighbor); + else + rsOut = getRowWiseSparsityVector(mb); + } + else { + switch(node.getOp()) { + case MM: + estimInternChain(node.getRight(), rsRightNeighbor, opRightNeighbor); + estimInternChain(node.getLeft(), (double[])(node.getRight().getSynopsis()), node.getOp()); + rsOut = (double[])node.getLeft().getSynopsis(); + break; + case CBIND: + /** NOTE: considering the current node as new DAG for estimation (cut), since the row sparsity of + * the right neighbor cannot be aggregated into a cbind operation when having only row sparsity vectors + */ + estimInternChain(node.getLeft()); + estimInternChain(node.getRight()); + double[] rsCBind = estimInternCBind((double[])(node.getLeft().getSynopsis()), (double[])(node.getRight().getSynopsis())); + if(rsRightNeighbor != null) { + rsOut = (double[])estimInternMMFallback(rsCBind, rsRightNeighbor); Review Comment: remove cast ########## src/main/java/org/apache/sysds/hops/estim/EstimatorRowWise.java: ########## @@ -0,0 +1,325 @@ +/* + * 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.sysds.hops.estim; + +import org.apache.commons.lang3.ArrayUtils; +import org.apache.commons.lang3.NotImplementedException; +import org.apache.sysds.hops.OptimizerUtils; +import org.apache.sysds.runtime.data.SparseRow; +import org.apache.sysds.runtime.matrix.data.MatrixBlock; +import org.apache.sysds.runtime.meta.DataCharacteristics; +import org.apache.sysds.runtime.meta.MatrixCharacteristics; + +import java.util.stream.DoubleStream; +import java.util.stream.IntStream; + +/** + * This estimator implements an approach based on row-wise sparsity estimation, + * introduced in + * Lin, Chunxu, Wensheng Luo, Yixiang Fang, Chenhao Ma, Xilin Liu and Yuchi Ma: + * On Efficient Large Sparse Matrix Chain Multiplication. + * Proceedings of the ACM on Management of Data 2 (2024): 1 - 27. + */ +public class EstimatorRowWise extends SparsityEstimator { + @Override + public DataCharacteristics estim(MMNode root) { + estimInternChain(root); + double sparsity = DoubleStream.of((double[])root.getSynopsis()).average().orElse(0); + + DataCharacteristics outputCharacteristics = deriveOutputCharacteristics(root, sparsity); + return root.setDataCharacteristics(outputCharacteristics); + } + + @Override + public double estim(MatrixBlock m1, MatrixBlock m2) { + return estim(m1, m2, OpCode.MM); + } + + @Override + public double estim(MatrixBlock m1, MatrixBlock m2, OpCode op) { + if( isExactMetadataOp(op, m1.getNumColumns()) ) { + return estimExactMetaData(m1.getDataCharacteristics(), + m2.getDataCharacteristics(), op).getSparsity(); + } + + double[] rsOut = estimIntern(m1, m2, op); + return DoubleStream.of(rsOut).average().orElse(0); + } + + @Override + public double estim(MatrixBlock m1, OpCode op) { + if( isExactMetadataOp(op, m1.getNumColumns()) ) + return estimExactMetaData(m1.getDataCharacteristics(), null, op).getSparsity(); + + double[] rsOut = estimIntern(m1, op); + return DoubleStream.of(rsOut).average().orElse(0); + } + + private void estimInternChain(MMNode node) { + estimInternChain(node, null, null); + } + + private void estimInternChain(MMNode node, double[] rsRightNeighbor, OpCode opRightNeighbor) { + double[] rsOut; + if(node.isLeaf()) { + MatrixBlock mb = node.getData(); + if(rsRightNeighbor != null) + rsOut = estimIntern(mb, rsRightNeighbor, opRightNeighbor); + else + rsOut = getRowWiseSparsityVector(mb); + } + else { + switch(node.getOp()) { + case MM: + estimInternChain(node.getRight(), rsRightNeighbor, opRightNeighbor); + estimInternChain(node.getLeft(), (double[])(node.getRight().getSynopsis()), node.getOp()); + rsOut = (double[])node.getLeft().getSynopsis(); + break; + case CBIND: + /** NOTE: considering the current node as new DAG for estimation (cut), since the row sparsity of + * the right neighbor cannot be aggregated into a cbind operation when having only row sparsity vectors + */ + estimInternChain(node.getLeft()); Review Comment: Move these .getLeft and .getRight to variables above the switch statement. ########## src/main/java/org/apache/sysds/hops/estim/EstimatorRowWise.java: ########## @@ -0,0 +1,325 @@ +/* + * 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.sysds.hops.estim; + +import org.apache.commons.lang3.ArrayUtils; +import org.apache.commons.lang3.NotImplementedException; +import org.apache.sysds.hops.OptimizerUtils; +import org.apache.sysds.runtime.data.SparseRow; +import org.apache.sysds.runtime.matrix.data.MatrixBlock; +import org.apache.sysds.runtime.meta.DataCharacteristics; +import org.apache.sysds.runtime.meta.MatrixCharacteristics; + +import java.util.stream.DoubleStream; +import java.util.stream.IntStream; + +/** + * This estimator implements an approach based on row-wise sparsity estimation, + * introduced in + * Lin, Chunxu, Wensheng Luo, Yixiang Fang, Chenhao Ma, Xilin Liu and Yuchi Ma: + * On Efficient Large Sparse Matrix Chain Multiplication. + * Proceedings of the ACM on Management of Data 2 (2024): 1 - 27. + */ +public class EstimatorRowWise extends SparsityEstimator { + @Override + public DataCharacteristics estim(MMNode root) { + estimInternChain(root); + double sparsity = DoubleStream.of((double[])root.getSynopsis()).average().orElse(0); + + DataCharacteristics outputCharacteristics = deriveOutputCharacteristics(root, sparsity); + return root.setDataCharacteristics(outputCharacteristics); + } + + @Override + public double estim(MatrixBlock m1, MatrixBlock m2) { + return estim(m1, m2, OpCode.MM); + } + + @Override + public double estim(MatrixBlock m1, MatrixBlock m2, OpCode op) { + if( isExactMetadataOp(op, m1.getNumColumns()) ) { + return estimExactMetaData(m1.getDataCharacteristics(), + m2.getDataCharacteristics(), op).getSparsity(); + } + + double[] rsOut = estimIntern(m1, m2, op); + return DoubleStream.of(rsOut).average().orElse(0); + } + + @Override + public double estim(MatrixBlock m1, OpCode op) { + if( isExactMetadataOp(op, m1.getNumColumns()) ) + return estimExactMetaData(m1.getDataCharacteristics(), null, op).getSparsity(); + + double[] rsOut = estimIntern(m1, op); + return DoubleStream.of(rsOut).average().orElse(0); + } + + private void estimInternChain(MMNode node) { + estimInternChain(node, null, null); + } + + private void estimInternChain(MMNode node, double[] rsRightNeighbor, OpCode opRightNeighbor) { + double[] rsOut; + if(node.isLeaf()) { + MatrixBlock mb = node.getData(); + if(rsRightNeighbor != null) + rsOut = estimIntern(mb, rsRightNeighbor, opRightNeighbor); + else + rsOut = getRowWiseSparsityVector(mb); + } + else { + switch(node.getOp()) { + case MM: + estimInternChain(node.getRight(), rsRightNeighbor, opRightNeighbor); + estimInternChain(node.getLeft(), (double[])(node.getRight().getSynopsis()), node.getOp()); + rsOut = (double[])node.getLeft().getSynopsis(); + break; + case CBIND: + /** NOTE: considering the current node as new DAG for estimation (cut), since the row sparsity of + * the right neighbor cannot be aggregated into a cbind operation when having only row sparsity vectors + */ + estimInternChain(node.getLeft()); + estimInternChain(node.getRight()); + double[] rsCBind = estimInternCBind((double[])(node.getLeft().getSynopsis()), (double[])(node.getRight().getSynopsis())); + if(rsRightNeighbor != null) { + rsOut = (double[])estimInternMMFallback(rsCBind, rsRightNeighbor); + if(opRightNeighbor != OpCode.MM) + throw new NotImplementedException("Fallback sparsity estimation has only been " + + "considered for MM operation w/ right neighbor yet."); + } + else + rsOut = (double[])rsCBind; + break; + case RBIND: + /** NOTE: considering the current node as new DAG for estimation (cut), since the row sparsity of + * the right neighbor cannot be aggregated into an rbind operation when having only row sparsity vectors + */ + estimInternChain(node.getLeft()); + estimInternChain(node.getRight()); + double[] rsRBind = estimInternRBind((double[])(node.getLeft().getSynopsis()), (double[])(node.getRight().getSynopsis())); + if(rsRightNeighbor != null) { + rsOut = (double[])estimInternMMFallback(rsRBind, rsRightNeighbor); + if(opRightNeighbor != OpCode.MM) + throw new NotImplementedException("Fallback sparsity estimation has only been " + + "considered for MM operation w/ right neighbor yet."); + } + else + rsOut = (double[])rsRBind; + break; + case PLUS: + /** NOTE: considering the current node as new DAG for estimation (cut), since the row sparsity of + * the right neighbor cannot be aggregated into an element-wise operation when having only row sparsity vectors + */ + estimInternChain(node.getLeft()); + estimInternChain(node.getRight()); + double[] rsPlus = estimInternPlus((double[])(node.getLeft().getSynopsis()), (double[])(node.getRight().getSynopsis())); + if(rsRightNeighbor != null) { + rsOut = (double[])estimInternMMFallback(rsPlus, rsRightNeighbor); + if(opRightNeighbor != OpCode.MM) + throw new NotImplementedException("Fallback sparsity estimation has only been " + + "considered for MM operation w/ right neighbor yet."); + } + else + rsOut = (double[])rsPlus; + break; + case MULT: + /** NOTE: considering the current node as new DAG for estimation (cut), since the row sparsity of + * the right neighbor cannot be aggregated into an element-wise operation when having only row sparsity vectors + */ + estimInternChain(node.getLeft()); + estimInternChain(node.getRight()); + double[] rsMult = estimInternMult((double[])(node.getLeft().getSynopsis()), (double[])(node.getRight().getSynopsis())); + if(rsRightNeighbor != null) { + rsOut = (double[])estimInternMMFallback(rsMult, rsRightNeighbor); + if(opRightNeighbor != OpCode.MM) + throw new NotImplementedException("Fallback sparsity estimation has only been " + + "considered for MM operation w/ right neighbor yet."); + } + else + rsOut = (double[])rsMult; + break; + default: + throw new NotImplementedException("Chain estimation for operator " + node.getOp().toString() + + " is not supported yet."); + } + } + node.setSynopsis(rsOut); + node.setDataCharacteristics(deriveOutputCharacteristics(node, DoubleStream.of(rsOut).average().orElse(0))); + return; + } + + private double[] estimIntern(MatrixBlock m1, MatrixBlock m2, OpCode op) { + double[] rsM2 = getRowWiseSparsityVector(m2); + return estimIntern(m1, rsM2, op); + } + + private double[] estimIntern(MatrixBlock m1, double[] rsM2, OpCode op) { + switch(op) { + case MM: + return estimInternMM(m1, rsM2); + case CBIND: + return estimInternCBind(getRowWiseSparsityVector(m1), rsM2); + case RBIND: + return estimInternRBind(getRowWiseSparsityVector(m1), rsM2); + case PLUS: + return estimInternPlus(getRowWiseSparsityVector(m1), rsM2); + case MULT: + return estimInternMult(getRowWiseSparsityVector(m1), rsM2); + default: + throw new NotImplementedException("Sparsity estimation for operation " + op.toString() + " not supported yet."); + } + } + + private double[] estimIntern(MatrixBlock mb, OpCode op) { + switch(op) { + case DIAG: + return estimInternDiag(mb); + default: + throw new NotImplementedException("Sparsity estimation for operation " + op.toString() + " not supported yet."); + } + } + + // Corresponds to Algorithm 1 in the publication + private double[] estimInternMM(MatrixBlock m1, double[] rsM2) { + double[] rsOut = IntStream.range(0, m1.getNumRows()).mapToDouble( + r -> (double) 1 - IntStream.of(getNonZeroColumnIndices(m1, r)).mapToDouble( + c -> (double) 1 - rsM2[c] + ).reduce((double) 1, (currentVal, val) -> currentVal * val)) + .toArray(); + return rsOut; + } + + // NOTE: this is the best estimation possible when we only have the two row sparsity vectors + private double[] estimInternMMFallback(double[] rsM1, double[] rsM2) { + // NOTE: Considering the average would probably not be far off while saving computing time + // double avgRsM2 = DoubleStream.of(rsM2).average().orElse(0); + // double[] rsOut = DoubleStream.of(rsM1).map( + // rsM1I -> (double) 1 - Math.pow((double) 1 - (rsM1I * avgRsM2), rsM2.length)).toArray(); Review Comment: Consider, formatting comments as docstrings above the method, or removing them. Do not leave commented code. ########## src/main/java/org/apache/sysds/hops/estim/EstimatorRowWise.java: ########## @@ -0,0 +1,325 @@ +/* + * 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.sysds.hops.estim; + +import org.apache.commons.lang3.ArrayUtils; +import org.apache.commons.lang3.NotImplementedException; +import org.apache.sysds.hops.OptimizerUtils; +import org.apache.sysds.runtime.data.SparseRow; +import org.apache.sysds.runtime.matrix.data.MatrixBlock; +import org.apache.sysds.runtime.meta.DataCharacteristics; +import org.apache.sysds.runtime.meta.MatrixCharacteristics; + +import java.util.stream.DoubleStream; +import java.util.stream.IntStream; + +/** + * This estimator implements an approach based on row-wise sparsity estimation, + * introduced in + * Lin, Chunxu, Wensheng Luo, Yixiang Fang, Chenhao Ma, Xilin Liu and Yuchi Ma: + * On Efficient Large Sparse Matrix Chain Multiplication. + * Proceedings of the ACM on Management of Data 2 (2024): 1 - 27. + */ +public class EstimatorRowWise extends SparsityEstimator { + @Override + public DataCharacteristics estim(MMNode root) { + estimInternChain(root); + double sparsity = DoubleStream.of((double[])root.getSynopsis()).average().orElse(0); + + DataCharacteristics outputCharacteristics = deriveOutputCharacteristics(root, sparsity); + return root.setDataCharacteristics(outputCharacteristics); + } + + @Override + public double estim(MatrixBlock m1, MatrixBlock m2) { + return estim(m1, m2, OpCode.MM); + } + + @Override + public double estim(MatrixBlock m1, MatrixBlock m2, OpCode op) { + if( isExactMetadataOp(op, m1.getNumColumns()) ) { + return estimExactMetaData(m1.getDataCharacteristics(), + m2.getDataCharacteristics(), op).getSparsity(); + } + + double[] rsOut = estimIntern(m1, m2, op); + return DoubleStream.of(rsOut).average().orElse(0); + } + + @Override + public double estim(MatrixBlock m1, OpCode op) { + if( isExactMetadataOp(op, m1.getNumColumns()) ) + return estimExactMetaData(m1.getDataCharacteristics(), null, op).getSparsity(); + + double[] rsOut = estimIntern(m1, op); + return DoubleStream.of(rsOut).average().orElse(0); + } + + private void estimInternChain(MMNode node) { + estimInternChain(node, null, null); + } + + private void estimInternChain(MMNode node, double[] rsRightNeighbor, OpCode opRightNeighbor) { + double[] rsOut; + if(node.isLeaf()) { + MatrixBlock mb = node.getData(); + if(rsRightNeighbor != null) + rsOut = estimIntern(mb, rsRightNeighbor, opRightNeighbor); + else + rsOut = getRowWiseSparsityVector(mb); + } + else { + switch(node.getOp()) { + case MM: + estimInternChain(node.getRight(), rsRightNeighbor, opRightNeighbor); + estimInternChain(node.getLeft(), (double[])(node.getRight().getSynopsis()), node.getOp()); + rsOut = (double[])node.getLeft().getSynopsis(); + break; + case CBIND: + /** NOTE: considering the current node as new DAG for estimation (cut), since the row sparsity of + * the right neighbor cannot be aggregated into a cbind operation when having only row sparsity vectors + */ + estimInternChain(node.getLeft()); + estimInternChain(node.getRight()); + double[] rsCBind = estimInternCBind((double[])(node.getLeft().getSynopsis()), (double[])(node.getRight().getSynopsis())); + if(rsRightNeighbor != null) { + rsOut = (double[])estimInternMMFallback(rsCBind, rsRightNeighbor); + if(opRightNeighbor != OpCode.MM) + throw new NotImplementedException("Fallback sparsity estimation has only been " + + "considered for MM operation w/ right neighbor yet."); + } + else + rsOut = (double[])rsCBind; + break; + case RBIND: + /** NOTE: considering the current node as new DAG for estimation (cut), since the row sparsity of + * the right neighbor cannot be aggregated into an rbind operation when having only row sparsity vectors + */ + estimInternChain(node.getLeft()); + estimInternChain(node.getRight()); + double[] rsRBind = estimInternRBind((double[])(node.getLeft().getSynopsis()), (double[])(node.getRight().getSynopsis())); + if(rsRightNeighbor != null) { + rsOut = (double[])estimInternMMFallback(rsRBind, rsRightNeighbor); + if(opRightNeighbor != OpCode.MM) + throw new NotImplementedException("Fallback sparsity estimation has only been " + + "considered for MM operation w/ right neighbor yet."); + } + else + rsOut = (double[])rsRBind; + break; + case PLUS: + /** NOTE: considering the current node as new DAG for estimation (cut), since the row sparsity of + * the right neighbor cannot be aggregated into an element-wise operation when having only row sparsity vectors + */ + estimInternChain(node.getLeft()); + estimInternChain(node.getRight()); + double[] rsPlus = estimInternPlus((double[])(node.getLeft().getSynopsis()), (double[])(node.getRight().getSynopsis())); + if(rsRightNeighbor != null) { + rsOut = (double[])estimInternMMFallback(rsPlus, rsRightNeighbor); + if(opRightNeighbor != OpCode.MM) + throw new NotImplementedException("Fallback sparsity estimation has only been " + + "considered for MM operation w/ right neighbor yet."); + } + else + rsOut = (double[])rsPlus; + break; + case MULT: + /** NOTE: considering the current node as new DAG for estimation (cut), since the row sparsity of + * the right neighbor cannot be aggregated into an element-wise operation when having only row sparsity vectors + */ + estimInternChain(node.getLeft()); + estimInternChain(node.getRight()); + double[] rsMult = estimInternMult((double[])(node.getLeft().getSynopsis()), (double[])(node.getRight().getSynopsis())); + if(rsRightNeighbor != null) { + rsOut = (double[])estimInternMMFallback(rsMult, rsRightNeighbor); + if(opRightNeighbor != OpCode.MM) + throw new NotImplementedException("Fallback sparsity estimation has only been " + + "considered for MM operation w/ right neighbor yet."); + } + else + rsOut = (double[])rsMult; + break; + default: + throw new NotImplementedException("Chain estimation for operator " + node.getOp().toString() + + " is not supported yet."); + } + } + node.setSynopsis(rsOut); + node.setDataCharacteristics(deriveOutputCharacteristics(node, DoubleStream.of(rsOut).average().orElse(0))); + return; + } + + private double[] estimIntern(MatrixBlock m1, MatrixBlock m2, OpCode op) { + double[] rsM2 = getRowWiseSparsityVector(m2); + return estimIntern(m1, rsM2, op); + } + + private double[] estimIntern(MatrixBlock m1, double[] rsM2, OpCode op) { + switch(op) { + case MM: + return estimInternMM(m1, rsM2); + case CBIND: + return estimInternCBind(getRowWiseSparsityVector(m1), rsM2); + case RBIND: + return estimInternRBind(getRowWiseSparsityVector(m1), rsM2); + case PLUS: + return estimInternPlus(getRowWiseSparsityVector(m1), rsM2); + case MULT: + return estimInternMult(getRowWiseSparsityVector(m1), rsM2); + default: + throw new NotImplementedException("Sparsity estimation for operation " + op.toString() + " not supported yet."); + } + } + + private double[] estimIntern(MatrixBlock mb, OpCode op) { + switch(op) { + case DIAG: + return estimInternDiag(mb); + default: + throw new NotImplementedException("Sparsity estimation for operation " + op.toString() + " not supported yet."); + } + } + + // Corresponds to Algorithm 1 in the publication + private double[] estimInternMM(MatrixBlock m1, double[] rsM2) { + double[] rsOut = IntStream.range(0, m1.getNumRows()).mapToDouble( + r -> (double) 1 - IntStream.of(getNonZeroColumnIndices(m1, r)).mapToDouble( + c -> (double) 1 - rsM2[c] + ).reduce((double) 1, (currentVal, val) -> currentVal * val)) + .toArray(); + return rsOut; + } + + // NOTE: this is the best estimation possible when we only have the two row sparsity vectors + private double[] estimInternMMFallback(double[] rsM1, double[] rsM2) { + // NOTE: Considering the average would probably not be far off while saving computing time + // double avgRsM2 = DoubleStream.of(rsM2).average().orElse(0); + // double[] rsOut = DoubleStream.of(rsM1).map( + // rsM1I -> (double) 1 - Math.pow((double) 1 - (rsM1I * avgRsM2), rsM2.length)).toArray(); + double[] rsOut = DoubleStream.of(rsM1).map( Review Comment: Im pretty sure the double stream here is slower than a normal explicit for loop. ########## src/main/java/org/apache/sysds/hops/estim/EstimatorRowWise.java: ########## @@ -0,0 +1,325 @@ +/* + * 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.sysds.hops.estim; + +import org.apache.commons.lang3.ArrayUtils; +import org.apache.commons.lang3.NotImplementedException; +import org.apache.sysds.hops.OptimizerUtils; +import org.apache.sysds.runtime.data.SparseRow; +import org.apache.sysds.runtime.matrix.data.MatrixBlock; +import org.apache.sysds.runtime.meta.DataCharacteristics; +import org.apache.sysds.runtime.meta.MatrixCharacteristics; + +import java.util.stream.DoubleStream; +import java.util.stream.IntStream; + +/** + * This estimator implements an approach based on row-wise sparsity estimation, + * introduced in + * Lin, Chunxu, Wensheng Luo, Yixiang Fang, Chenhao Ma, Xilin Liu and Yuchi Ma: + * On Efficient Large Sparse Matrix Chain Multiplication. + * Proceedings of the ACM on Management of Data 2 (2024): 1 - 27. + */ +public class EstimatorRowWise extends SparsityEstimator { + @Override + public DataCharacteristics estim(MMNode root) { + estimInternChain(root); + double sparsity = DoubleStream.of((double[])root.getSynopsis()).average().orElse(0); + + DataCharacteristics outputCharacteristics = deriveOutputCharacteristics(root, sparsity); + return root.setDataCharacteristics(outputCharacteristics); + } + + @Override + public double estim(MatrixBlock m1, MatrixBlock m2) { + return estim(m1, m2, OpCode.MM); + } + + @Override + public double estim(MatrixBlock m1, MatrixBlock m2, OpCode op) { + if( isExactMetadataOp(op, m1.getNumColumns()) ) { + return estimExactMetaData(m1.getDataCharacteristics(), + m2.getDataCharacteristics(), op).getSparsity(); + } + + double[] rsOut = estimIntern(m1, m2, op); + return DoubleStream.of(rsOut).average().orElse(0); + } + + @Override + public double estim(MatrixBlock m1, OpCode op) { + if( isExactMetadataOp(op, m1.getNumColumns()) ) + return estimExactMetaData(m1.getDataCharacteristics(), null, op).getSparsity(); + + double[] rsOut = estimIntern(m1, op); + return DoubleStream.of(rsOut).average().orElse(0); + } + + private void estimInternChain(MMNode node) { + estimInternChain(node, null, null); + } + + private void estimInternChain(MMNode node, double[] rsRightNeighbor, OpCode opRightNeighbor) { + double[] rsOut; + if(node.isLeaf()) { + MatrixBlock mb = node.getData(); + if(rsRightNeighbor != null) + rsOut = estimIntern(mb, rsRightNeighbor, opRightNeighbor); + else + rsOut = getRowWiseSparsityVector(mb); + } + else { + switch(node.getOp()) { + case MM: + estimInternChain(node.getRight(), rsRightNeighbor, opRightNeighbor); + estimInternChain(node.getLeft(), (double[])(node.getRight().getSynopsis()), node.getOp()); + rsOut = (double[])node.getLeft().getSynopsis(); + break; + case CBIND: + /** NOTE: considering the current node as new DAG for estimation (cut), since the row sparsity of + * the right neighbor cannot be aggregated into a cbind operation when having only row sparsity vectors + */ + estimInternChain(node.getLeft()); + estimInternChain(node.getRight()); + double[] rsCBind = estimInternCBind((double[])(node.getLeft().getSynopsis()), (double[])(node.getRight().getSynopsis())); + if(rsRightNeighbor != null) { + rsOut = (double[])estimInternMMFallback(rsCBind, rsRightNeighbor); + if(opRightNeighbor != OpCode.MM) + throw new NotImplementedException("Fallback sparsity estimation has only been " + + "considered for MM operation w/ right neighbor yet."); + } + else + rsOut = (double[])rsCBind; + break; + case RBIND: + /** NOTE: considering the current node as new DAG for estimation (cut), since the row sparsity of + * the right neighbor cannot be aggregated into an rbind operation when having only row sparsity vectors + */ + estimInternChain(node.getLeft()); + estimInternChain(node.getRight()); + double[] rsRBind = estimInternRBind((double[])(node.getLeft().getSynopsis()), (double[])(node.getRight().getSynopsis())); + if(rsRightNeighbor != null) { + rsOut = (double[])estimInternMMFallback(rsRBind, rsRightNeighbor); + if(opRightNeighbor != OpCode.MM) + throw new NotImplementedException("Fallback sparsity estimation has only been " + + "considered for MM operation w/ right neighbor yet."); + } + else + rsOut = (double[])rsRBind; + break; + case PLUS: + /** NOTE: considering the current node as new DAG for estimation (cut), since the row sparsity of + * the right neighbor cannot be aggregated into an element-wise operation when having only row sparsity vectors + */ + estimInternChain(node.getLeft()); + estimInternChain(node.getRight()); + double[] rsPlus = estimInternPlus((double[])(node.getLeft().getSynopsis()), (double[])(node.getRight().getSynopsis())); + if(rsRightNeighbor != null) { + rsOut = (double[])estimInternMMFallback(rsPlus, rsRightNeighbor); + if(opRightNeighbor != OpCode.MM) + throw new NotImplementedException("Fallback sparsity estimation has only been " + + "considered for MM operation w/ right neighbor yet."); + } + else + rsOut = (double[])rsPlus; + break; + case MULT: + /** NOTE: considering the current node as new DAG for estimation (cut), since the row sparsity of + * the right neighbor cannot be aggregated into an element-wise operation when having only row sparsity vectors + */ + estimInternChain(node.getLeft()); + estimInternChain(node.getRight()); + double[] rsMult = estimInternMult((double[])(node.getLeft().getSynopsis()), (double[])(node.getRight().getSynopsis())); + if(rsRightNeighbor != null) { + rsOut = (double[])estimInternMMFallback(rsMult, rsRightNeighbor); + if(opRightNeighbor != OpCode.MM) + throw new NotImplementedException("Fallback sparsity estimation has only been " + + "considered for MM operation w/ right neighbor yet."); + } + else + rsOut = (double[])rsMult; + break; + default: + throw new NotImplementedException("Chain estimation for operator " + node.getOp().toString() + + " is not supported yet."); + } + } + node.setSynopsis(rsOut); + node.setDataCharacteristics(deriveOutputCharacteristics(node, DoubleStream.of(rsOut).average().orElse(0))); + return; + } + + private double[] estimIntern(MatrixBlock m1, MatrixBlock m2, OpCode op) { + double[] rsM2 = getRowWiseSparsityVector(m2); + return estimIntern(m1, rsM2, op); + } + + private double[] estimIntern(MatrixBlock m1, double[] rsM2, OpCode op) { + switch(op) { + case MM: + return estimInternMM(m1, rsM2); + case CBIND: + return estimInternCBind(getRowWiseSparsityVector(m1), rsM2); + case RBIND: + return estimInternRBind(getRowWiseSparsityVector(m1), rsM2); + case PLUS: + return estimInternPlus(getRowWiseSparsityVector(m1), rsM2); + case MULT: + return estimInternMult(getRowWiseSparsityVector(m1), rsM2); + default: + throw new NotImplementedException("Sparsity estimation for operation " + op.toString() + " not supported yet."); + } + } + + private double[] estimIntern(MatrixBlock mb, OpCode op) { + switch(op) { + case DIAG: + return estimInternDiag(mb); + default: + throw new NotImplementedException("Sparsity estimation for operation " + op.toString() + " not supported yet."); + } + } + + // Corresponds to Algorithm 1 in the publication + private double[] estimInternMM(MatrixBlock m1, double[] rsM2) { + double[] rsOut = IntStream.range(0, m1.getNumRows()).mapToDouble( + r -> (double) 1 - IntStream.of(getNonZeroColumnIndices(m1, r)).mapToDouble( + c -> (double) 1 - rsM2[c] + ).reduce((double) 1, (currentVal, val) -> currentVal * val)) + .toArray(); + return rsOut; + } + + // NOTE: this is the best estimation possible when we only have the two row sparsity vectors + private double[] estimInternMMFallback(double[] rsM1, double[] rsM2) { + // NOTE: Considering the average would probably not be far off while saving computing time + // double avgRsM2 = DoubleStream.of(rsM2).average().orElse(0); + // double[] rsOut = DoubleStream.of(rsM1).map( + // rsM1I -> (double) 1 - Math.pow((double) 1 - (rsM1I * avgRsM2), rsM2.length)).toArray(); + double[] rsOut = DoubleStream.of(rsM1).map( + rsM1I -> (double) 1 - DoubleStream.of(rsM2).reduce((double) 1, + (currentVal, rsM2J) -> currentVal * ((double) 1 - (rsM1I * rsM2J)))).toArray(); + return rsOut; + } + + private double[] estimInternCBind(double[] rsM1, double[] rsM2) { + // FIXME: this assumes that the number of columns is equivalent for both inputs + return IntStream.range(0, rsM1.length).mapToDouble( + idx -> (rsM1[idx] + rsM2[idx]) / (double) 2).toArray(); Review Comment: try to time this function and compare it to: ``` double[] ret = new double[rsM1.length]; for(int i = 0; i < rsM1.length; i++){ ret[i] = (rsM1[i] + rsM2[i]) / 2.0; } return ret; ``` -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
