[FLINK-27366] Record schema on filesystem path #101
Conversation
JingsongLi
changed the title
JingsongLi
force-pushed
the
schema
branch
3 times, most recently
from
2e1c59d to
c7c9095
Compare
| /** Json serializer for jackson. */ | ||
| public interface JsonSerializer<T> { | ||
|
|
||
| void serializer(T t, JsonGenerator generator) throws IOException; |
|
|
||
| void serializer(T t, JsonGenerator generator) throws IOException; | ||
|
|
||
| T deserializer(JsonNode node); |
| return fieldIds.stream().max(Integer::compareTo).orElse(-1); | ||
| } | ||
|
|
||
| private static void collectFieldIds(Set<Integer> fieldIds, DataType type) { |
There was a problem hiding this comment.
DataType type, Set<Integer> fieldIds. Input arguments should be in front of output arguments.
| return listVersionedFiles(schemaDirectory(), SCHEMA_PREFIX) | ||
| .reduce(Math::max) | ||
| .map(this::schema); |
There was a problem hiding this comment.
Add hint files just like SnapshotFinder? Maybe extract common classes for both snapshot and schema.
There was a problem hiding this comment.
I don't think it's necessary for schema because there won't be too many versions of schema
There was a problem hiding this comment.
schema changes are a very low frequency thing (Compared to snapshot generation)
| return schema; | ||
| } else { | ||
| // retry | ||
| FileUtils.deleteOrWarn(temp); |
There was a problem hiding this comment.
If this method fails with exception temp file will not be cleaned.
There was a problem hiding this comment.
Yes, but we dont have other solutions.
There was a problem hiding this comment.
You have. Wrap the code with try... finally....
| import static org.junit.jupiter.api.Assertions.assertThrows; | ||
|
|
||
| /** Test for {@link SchemaManager}. */ | ||
| public class SchemaManagerTest { |
There was a problem hiding this comment.
Lacks concurrent tests and cleanup tests for commitNewVersion.
There was a problem hiding this comment.
I'll add after each check for cleanup tests.
There was a problem hiding this comment.
I'll add one testConcurrentCommit test.
| FileUtils.writeFileUtf8(temp, schema.toString()); | ||
|
|
||
| Boolean success = lock.runWithLock(() -> temp.getFileSystem().rename(temp, finalFile)); | ||
| if (success) { | ||
| return schema; | ||
| } else { | ||
| // retry | ||
| FileUtils.deleteOrWarn(temp); | ||
| boolean success = false; | ||
| try { |
There was a problem hiding this comment.
try should also cover file write. File write may be partial
2 participants
We can store the schema on the path of the table store, which includes type, options, etc.
This schema should be in a format that supports evolution, which means that the fields contain id information.