Merge pull request #72 from r-uehara0219/feature/varint-length-field

Use varint length field for last_path encoding to support longer GCP object names

Author: dmikurube

src/main/java/org/embulk/input/gcs/GcsFileInput.java

@@ -91,28 +91,50 @@ static FileList listFiles(final PluginTask task) {
 
     // String nextToken = base64Encode(0x0a + ASCII character according to utf8EncodeLength position+ filePath);
     static String base64Encode(final String path) {
-        byte[] encoding;
         byte[] utf8 = path.getBytes(StandardCharsets.UTF_8);
         LOG.debug("path string: {} ,path length:{} \" + ", path, utf8.length);
 
         int utf8EncodeLength = utf8.length;
-        if (utf8EncodeLength >= 128) {
-            throw new ConfigException(String.format("last_path '%s' is too long to encode. Please try to reduce its length", path));
+        // GCP object names can be up to 1024 bytes in length.
+        // This limit aligns with task.getLastPath() expectations.
+        if (utf8EncodeLength >= 1025) {
+            throw new ConfigException(String.format("last_path '%s' is too long to encode. Maximum allowed is 1024 bytes", path));
         }
 
-        encoding = new byte[utf8.length + 2];
+        byte[] lengthVarint;
+        byte[] encoding;
+        lengthVarint = encodeVarint(utf8EncodeLength);
+        encoding = new byte[1 + lengthVarint.length + utf8.length];
         encoding[0] = 0x0a;
 
-        // for example: 60 -> '<'
-        char temp = (char) utf8EncodeLength;
-        encoding[1] = (byte) temp;
-        System.arraycopy(utf8, 0, encoding, 2, utf8.length);
+        System.arraycopy(lengthVarint, 0, encoding, 1, lengthVarint.length);
+        System.arraycopy(utf8, 0, encoding, 1 + lengthVarint.length, utf8.length);
 
         final String s = Base64.getEncoder().encodeToString(encoding);
         LOG.debug("last_path(base64 encoded): {}", s);
         return s;
     }
 
+    // see: https://protobuf.dev/programming-guides/encoding/#varints
+    static byte[] encodeVarint(int value) {
+        // utf8EncodeLength.length is up to 65535, so 2 bytes are enough for buffer
+        byte[] buffer = new byte[2];
+        int pos = 0;
+        while (true) {
+            int bits = value & 0x7F;
+            value >>>= 7;
+            if (value != 0) {
+                buffer[pos++] = (byte) (bits | 0x80);
+            } else {
+                buffer[pos++] = (byte) bits;
+                break;
+            }
+        }
+        byte[] result = new byte[pos];
+        System.arraycopy(buffer, 0, result, 0, pos);
+        return result;
+    }
+
     private static void printBucketInfo(final Storage client, final String bucket) {
         // get Bucket
         Storage.BucketGetOption fields = Storage.BucketGetOption.fields(

src/main/java/org/embulk/input/gcs/GcsFileInputPlugin.java

@@ -17,6 +17,7 @@
 package org.embulk.input.gcs;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Optional;
 import org.embulk.config.ConfigDiff;
@@ -64,10 +65,10 @@ public ConfigDiff transaction(final ConfigSource config, final FileInputPlugin.C
             }
         }
 
-        // @see https://cloud.google.com/storage/docs/bucket-naming
+        // @see https://cloud.google.com/storage/docs/objects#naming
         if (task.getLastPath().isPresent()) {
-            if (task.getLastPath().get().length() >= 128) {
-                throw new ConfigException("last_path length is allowed up to 127 characters");
+            if (task.getLastPath().get().getBytes(StandardCharsets.UTF_8).length >= 1025) {
+                throw new ConfigException("last_path is too long, which can contain a maximum of 1024 bytes encoded in UTF-8.");
             }
         }
 

src/test/java/org/embulk/input/gcs/TestGcsFileInputPlugin.java

@@ -18,6 +18,7 @@
 
 import static org.embulk.input.gcs.GcsFileInputPlugin.CONFIG_MAPPER;
 import static org.embulk.input.gcs.GcsFileInputPlugin.CONFIG_MAPPER_FACTORY;
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assume.assumeNotNull;
@@ -394,15 +395,34 @@ public void testBase64() {
         assertEquals("CgJjMg==", GcsFileInput.base64Encode("c2"));
         assertEquals("Cgh0ZXN0LmNzdg==", GcsFileInput.base64Encode("test.csv"));
         assertEquals("ChZnY3MtdGVzdC9zYW1wbGVfMDEuY3N2", GcsFileInput.base64Encode("gcs-test/sample_01.csv"));
-        String params = "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc127";
-        String expected = "Cn9jY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjMTI3";
+        String params = "ccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc200";
+        String expected = "CsgBY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2NjY2MyMDA=";
         assertEquals(expected, GcsFileInput.base64Encode(params));
 
         params = "テストダミー/テス123/テストダミー/テストダミ.csv";
         expected = "CkPjg4bjgrnjg4jjg4Djg5/jg7wv44OG44K5MTIzL+ODhuOCueODiOODgOODn+ODvC/jg4bjgrnjg4jjg4Djg58uY3N2";
         assertEquals(expected, GcsFileInput.base64Encode(params));
     }
 
+    @Test
+    public void testEncodeVarint() {
+        byte[] expected1 = new byte[]{0x01};
+        byte[] result1 = GcsFileInput.encodeVarint(1);
+        assertArrayEquals("encodeVarint(1) should return {0x01}", expected1, result1);
+
+        byte[] expected127 = new byte[]{0x7F};
+        byte[] result127 = GcsFileInput.encodeVarint(127);
+        assertArrayEquals("encodeVarint(127) should return {0x7F}", expected127, result127);
+
+        byte[] expected128 = new byte[]{(byte) 0x80, 0x01};
+        byte[] result128 = GcsFileInput.encodeVarint(128);
+        assertArrayEquals("encodeVarint(128) should return {0x80, 0x01}", expected128, result128);
+
+        byte[] expected1024 = new byte[]{(byte) 0x80, 0x08};
+        byte[] result1024 = GcsFileInput.encodeVarint(1024);
+        assertArrayEquals("encodeVarint(1024) should return {0x80, 0x08}", expected1024, result1024);
+    }
+
     private ConfigSource config() {
         ConfigSource config = CONFIG_MAPPER_FACTORY.newConfigSource()
                 .set("bucket", GCP_BUCKET)