davidak09 commented on a change in pull request #12050:
URL: https://github.com/apache/beam/pull/12050#discussion_r445485378
##########
File path:
sdks/java/core/src/main/java/org/apache/beam/sdk/io/DefaultFilenamePolicy.java
##########
@@ -182,19 +184,26 @@ public void encode(Params value, OutputStream outStream)
throws IOException {
if (value == null) {
throw new CoderException("cannot encode a null value");
}
- stringCoder.encode(value.baseFilename.get().toString(), outStream);
- stringCoder.encode(value.shardTemplate, outStream);
- stringCoder.encode(value.suffix, outStream);
+ STRING_CODER.encode(value.baseFilename.get().toString(), outStream);
+ STRING_CODER.encode(value.shardTemplate, outStream);
+ STRING_CODER.encode(value.suffix, outStream);
+ BOOLEAN_CODER.encode(value.baseFilename.get().isDirectory(), outStream);
}
@Override
public Params decode(InputStream inStream) throws IOException {
- ResourceId prefix =
-
FileBasedSink.convertToFileResourceIfPossible(stringCoder.decode(inStream));
- String shardTemplate = stringCoder.decode(inStream);
- String suffix = stringCoder.decode(inStream);
+ String prefix = STRING_CODER.decode(inStream);
+ String shardTemplate = STRING_CODER.decode(inStream);
+ String suffix = STRING_CODER.decode(inStream);
+ ResourceId baseFilename;
+ if (inStream.available() > 0) {
+ baseFilename = FileSystems.matchNewResource(prefix,
BOOLEAN_CODER.decode(inStream));
+ } else {
+ // fallback for ensure backward compatibility
+ baseFilename = FileBasedSink.convertToFileResourceIfPossible(prefix);
Review comment:
Exactly, that's the issue, for example `LocalFileSystem` matches path
`/tmp/dirA/` as file..
How about something like this?
```java
public static class ParamsCoder extends AtomicCoder<Params> {
private static final ParamsCoder INSTANCE = new ParamsCoder();
private static final Coder<String> STRING_CODER = StringUtf8Coder.of();
public static ParamsCoder of() {
return INSTANCE;
}
@Override
public void encode(Params value, OutputStream outStream) throws
IOException {
if (value == null) {
throw new CoderException("cannot encode a null value");
}
ResourceId baseFilename = value.getBaseFilename().get();
String baseFilenameString =
(baseFilename.isDirectory() &&
!baseFilename.toString().endsWith("/"))
? baseFilename.toString() + "/"
: baseFilename.toString();
STRING_CODER.encode(baseFilenameString, outStream);
STRING_CODER.encode(value.getShardTemplate(), outStream);
STRING_CODER.encode(value.getSuffix(), outStream);
}
@Override
public Params decode(InputStream inStream) throws IOException {
String baseFilenameString = STRING_CODER.decode(inStream);
ResourceId prefix =
FileSystems.matchNewResource(baseFilenameString,
baseFilenameString.endsWith("/"));
String shardTemplate = STRING_CODER.decode(inStream);
String suffix = STRING_CODER.decode(inStream);
return new Params()
.withBaseFilename(prefix)
.withShardTemplate(shardTemplate)
.withSuffix(suffix);
}
}
```
But I'm not sure how to handle the suffix (`/`) to ensure compatibility
among (file) systems
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
