ywcb00 commented on code in PR #2466: URL: https://github.com/apache/systemds/pull/2466#discussion_r3271914964
########## src/main/java/org/apache/sysds/hops/estim/EstimatorRowWise.java: ########## @@ -0,0 +1,349 @@ +/* + * 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 double[] estimInternChain(MMNode node) { + return estimInternChain(node, null, null); + } + + private double[] 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 { + MMNode nodeLeft = node.getLeft(); + MMNode nodeRight = node.getRight(); + switch(node.getOp()) { + case MM: + double[] rsRightMM = estimInternChain(nodeRight, rsRightNeighbor, opRightNeighbor); + rsOut = estimInternChain(nodeLeft, rsRightMM, node.getOp()); + 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 + */ + double[] rsLeftCBind = estimInternChain(nodeLeft); + double[] rsRightCBind = estimInternChain(nodeRight); + double[] rsCBind = estimInternCBind(rsLeftCBind, rsRightCBind); + if(rsRightNeighbor != null) { + rsOut = 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 = 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 + */ + double[] rsLeftRBind = estimInternChain(nodeLeft); + double[] rsRightRBind = estimInternChain(nodeRight); + double[] rsRBind = estimInternRBind(rsLeftRBind, rsRightRBind); + if(rsRightNeighbor != null) { + rsOut = 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 = 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 + */ + double[] rsLeftPlus = estimInternChain(nodeLeft); + double[] rsRightPlus = estimInternChain(nodeRight); + double[] rsPlus = estimInternPlus(rsLeftPlus, rsRightPlus); + if(rsRightNeighbor != null) { + rsOut = 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 = 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 + */ + double[] rsLeftMult = estimInternChain(nodeLeft); + double[] rsRightMult = estimInternChain(nodeRight); + double[] rsMult = estimInternMult(rsLeftMult, rsRightMult); + if(rsRightNeighbor != null) { + rsOut = 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 = 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 rsOut; + } + + 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 = new double[m1.getNumRows()]; + for(int rIdx = 0; rIdx < m1.getNumRows(); rIdx++) { + double currentVal = 1; + for(int cIdx : getNonZeroColumnIndices(m1, rIdx)) { + currentVal *= (double) 1 - rsM2[cIdx]; + } + rsOut[rIdx] = 1 - currentVal; + } + return rsOut; + } + + /** + * NOTE: this is the best estimation possible when we only have the two row sparsity vectors + * NOTE: Considering the average of the second matrix would probably not be far off while saving computing time + */ + private double[] estimInternMMFallback(double[] rsM1, double[] rsM2) { Review Comment: The size of the output depends on the size of the first input. However, it is independent of the size of the second input and does not take the number of columns into account, as it is uses only row sparsity vectors. -- 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]
