[SYSTEMDS-3650] New I/O Support Cloud-optimized GeoTIFF

DIA WiSe 24/25 project
Closes #2195.

Co-authored-by: lorenzkautzsch <[email protected]>
Co-authored-by: Aditya Pandey <[email protected]>

Author: 3 people

docs/site/dml-language-reference.md

@@ -868,13 +868,15 @@ that users provide MTD files for their own data as well.
 
 #### File formats and MTD files
 
-SystemDS supports 4 file formats:
+SystemDS supports 6 file formats:
 
   * CSV (delimited)
   * Matrix Market (coordinate)
   * Text (i,j,v)
+  * LIBSVM
   * Binary
   * HDF5
+  * COG
 
 The CSV format is a standard text-based format where columns are separated by delimiter characters, typically commas, and
 rows are represented on separate lines.
@@ -899,6 +901,15 @@ Hierarchical Data Format (HDF) is a file format designed to store and organize l
 some features of the HDF5 like two dimension data (Matrix), matrix with FP64 (double) data type, 
 single dataset, single group, and contiguous dataset.
 
+Cloud Optimized GeoTIFF (COG) is an image format designed to store large amounts of geospatial data while allowing for
+efficient access. This is done by splitting the image into tiles which can then be accessed independently. Currently, SystemDS
+only supports reading COG files and can only process the most important metadata that is required for reading the image. Normal
+TIFF files that aren't tiled cannot be read by SystemDS. Support for BigTIFF is very limited and not recommended.
+
+The currently supported compression methods are as follows:
+- None
+- Deflate
+
 Let's look at a matrix and examples of its data represented in the supported formats with corresponding metadata. In the table below, we have
 a matrix consisting of 4 rows and 3 columns.
 

src/main/java/org/apache/sysds/common/Types.java

@@ -863,6 +863,7 @@ public enum FileFormat {
 		FEDERATED, // A federated matrix
 		PROTO,  // protocol buffer representation
 		HDF5,   // Hierarchical Data Format (HDF)
+		COG,   // Cloud-optimized GeoTIFF
 		UNKNOWN;
 
 		public boolean isIJV() {

src/main/java/org/apache/sysds/lops/compile/Dag.java

@@ -30,7 +30,6 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.sysds.api.DMLScript;
-import org.apache.sysds.common.Opcodes;
 import org.apache.sysds.common.Types.DataType;
 import org.apache.sysds.common.Types.ExecType;
 import org.apache.sysds.common.Types.FileFormat;

src/main/java/org/apache/sysds/parser/DMLTranslator.java

@@ -32,7 +32,6 @@
 import org.apache.commons.logging.LogFactory;
 import org.apache.sysds.api.DMLScript;
 import org.apache.sysds.common.Builtins;
-import org.apache.sysds.common.Opcodes;
 import org.apache.sysds.common.Types.AggOp;
 import org.apache.sysds.common.Types.DataType;
 import org.apache.sysds.common.Types.Direction;

src/main/java/org/apache/sysds/parser/DataExpression.java

@@ -1164,6 +1164,8 @@ else if( getVarParam(READNNZPARAM) != null ) {
 
 			boolean isHDF5 = (formatTypeString != null && formatTypeString.equalsIgnoreCase(FileFormat.HDF5.toString()));
 
+			boolean isCOG = (formatTypeString != null && formatTypeString.equalsIgnoreCase(FileFormat.COG.toString()));
+
 			dataTypeString = (getVarParam(DATATYPEPARAM) == null) ? null : getVarParam(DATATYPEPARAM).toString();
 
 			if ( dataTypeString == null || dataTypeString.equalsIgnoreCase(Statement.MATRIX_DATA_TYPE) 
@@ -1188,7 +1190,7 @@ else if( getVarParam(READNNZPARAM) != null ) {
 				// initialize size of target data identifier to UNKNOWN
 				getOutput().setDimensions(-1, -1);
 
-				if (!isCSV && !isLIBSVM && !isHDF5 && ConfigurationManager.getCompilerConfig()
+				if (!isCSV && !isLIBSVM && !isHDF5 && !isCOG && ConfigurationManager.getCompilerConfig()
 						.getBool(ConfigType.REJECT_READ_WRITE_UNKNOWNS) //skip check for csv/libsvm format / jmlc api
 					&& (getVarParam(READROWPARAM) == null || getVarParam(READCOLPARAM) == null) ) {
 						raiseValidateError("Missing or incomplete dimension information in read statement: "

src/main/java/org/apache/sysds/runtime/compress/colgroup/dictionary/ACachingMBDictionary.java

@@ -23,6 +23,7 @@
 
 public abstract class ACachingMBDictionary extends ADictionary {
 
+	private static final long serialVersionUID = 7035552219254994595L;
 	/** A Cache to contain a materialized version of the identity matrix. */
 	protected volatile SoftReference<MatrixBlockDictionary> cache = null;
 

src/main/java/org/apache/sysds/runtime/compress/colgroup/dictionary/AIdentityDictionary.java

@@ -22,6 +22,7 @@
 import org.apache.sysds.runtime.compress.DMLCompressionException;
 
 public abstract class AIdentityDictionary extends ACachingMBDictionary {
+	private static final long serialVersionUID = 5013713435287705877L;
 	/** The number of rows or columns, rows can be +1 if withEmpty is set. */
 	protected final int nRowCol;
 	/** Specify if the Identity matrix should contain an empty row in the end. */

src/main/java/org/apache/sysds/runtime/functionobjects/ParameterizedBuiltin.java

@@ -27,7 +27,6 @@
 import org.apache.commons.math3.distribution.FDistribution;
 import org.apache.commons.math3.distribution.NormalDistribution;
 import org.apache.commons.math3.distribution.TDistribution;
-import org.apache.sysds.common.Opcodes;
 import org.apache.sysds.runtime.DMLRuntimeException;
 import org.apache.sysds.runtime.util.UtilFunctions;
 

src/main/java/org/apache/sysds/runtime/instructions/gpu/BuiltinBinaryGPUInstruction.java

@@ -19,7 +19,6 @@
 
 package org.apache.sysds.runtime.instructions.gpu;
 
-import org.apache.sysds.common.Opcodes;
 import org.apache.sysds.common.Types.DataType;
 import org.apache.sysds.common.Types.ValueType;
 import org.apache.sysds.runtime.DMLRuntimeException;
@@ -77,5 +76,4 @@ else if (isMatrixScalar && (opcode.equals("min") || opcode.equals("max")))
 			throw new DMLRuntimeException(
 				"GPU : Unsupported GPU builtin operations on a matrix and a scalar:" + opcode);
 	}
-
 }

src/main/java/org/apache/sysds/runtime/instructions/gpu/MatrixBuiltinGPUInstruction.java

@@ -21,7 +21,6 @@
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.apache.sysds.common.Opcodes;
 import org.apache.sysds.runtime.DMLRuntimeException;
 import org.apache.sysds.runtime.controlprogram.caching.MatrixObject;
 import org.apache.sysds.runtime.controlprogram.context.ExecutionContext;

src/main/java/org/apache/sysds/runtime/io/FileFormatPropertiesCOG.java

@@ -0,0 +1,50 @@
+/*
+ * 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.runtime.io;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import java.io.Serializable;
+
+public class FileFormatPropertiesCOG extends FileFormatProperties implements Serializable {
+	protected static final Log LOG = LogFactory.getLog(FileFormatPropertiesCOG.class.getName());
+	private static final long serialVersionUID = 1038419221722594985L;
+
+	private String datasetName;
+
+	public FileFormatPropertiesCOG() {
+		this.datasetName = "systemdscog";
+	}
+
+	public FileFormatPropertiesCOG(String datasetName) {
+		this.datasetName = datasetName;
+	}
+
+	public String getDatasetName() {
+		return datasetName;
+	}
+
+	@Override public String toString() {
+		StringBuilder sb = new StringBuilder();
+		sb.append(" datasetName " + datasetName);
+		return sb.toString();
+	}
+}

src/main/java/org/apache/sysds/runtime/io/MatrixReaderFactory.java

@@ -64,8 +64,15 @@ public static MatrixReader createMatrixReader(FileFormat fmt) {
 				break;
 
 			case HDF5:
-				reader = (par & mcsr) ? new ReaderHDF5Parallel(
-					new FileFormatPropertiesHDF5()) : new ReaderHDF5(new FileFormatPropertiesHDF5());
+				reader = (par & mcsr) ? 
+					new ReaderHDF5Parallel(new FileFormatPropertiesHDF5()) : 
+					new ReaderHDF5(new FileFormatPropertiesHDF5());
+				break;
+
+			case COG:
+				reader = (par & mcsr) ?
+					new ReaderCOGParallel(new FileFormatPropertiesCOG()) :
+					new ReaderCOG(new FileFormatPropertiesCOG());
 				break;
 
 			case COMPRESSED:
@@ -124,6 +131,12 @@ public static MatrixReader createMatrixReader( ReadProperties props )  {
 					fileFormatPropertiesHDF5);
 				break;
 
+			case COG:
+				FileFormatPropertiesCOG fileFormatPropertiesCOG = props.formatProperties != null ? (FileFormatPropertiesCOG) props.formatProperties : new FileFormatPropertiesCOG();
+				reader = (par & mcsr) ?
+						new ReaderCOGParallel(fileFormatPropertiesCOG) : new ReaderCOG(fileFormatPropertiesCOG);
+				break;
+
 			case COMPRESSED:
 				reader = new ReaderCompressed();
 				break;