-
Notifications
You must be signed in to change notification settings - Fork 1.1k
[IOTDB-5]fundamental classes for supporting deletion #6
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 6 commits
d072e9e
dbcf2ed
ebbf9f8
4740d95
0cb7a98
2ffc23a
228f614
014aa86
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| /** | ||
| * Copyright © 2019 Apache IoTDB(incubating) (dev@iotdb.apache.org) | ||
| * <p> | ||
| * Licensed 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 | ||
| * <p> | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * <p> | ||
| * 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.iotdb.db.engine.modification; | ||
|
|
||
| /** | ||
| * Deletion is a delete operation on a timeseries. | ||
| */ | ||
| public class Deletion extends Modification { | ||
| private long timestamp; | ||
|
|
||
| public Deletion(String path, long versionNum, long timestamp) { | ||
| super(Type.DELETION, path, versionNum); | ||
| this.timestamp = timestamp; | ||
| } | ||
|
|
||
| public long getTimestamp() { | ||
| return timestamp; | ||
| } | ||
|
|
||
| public void setTimestamp(long timestamp) { | ||
| this.timestamp = timestamp; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean equals(Object obj) { | ||
| if (!(obj instanceof Modification)) | ||
| return false; | ||
| Deletion del = (Deletion) obj; | ||
| return super.equals(obj) && del.timestamp == this.timestamp; | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| /** | ||
| * Copyright © 2019 Apache IoTDB(incubating) (dev@iotdb.apache.org) | ||
|
jixuan1989 marked this conversation as resolved.
Outdated
|
||
| * <p> | ||
| * Licensed 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 | ||
| * <p> | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * <p> | ||
| * 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.iotdb.db.engine.modification; | ||
|
|
||
| /** | ||
| * Modification represents an UPDATE or DELETE operation on a certain timeseries. | ||
| */ | ||
| public abstract class Modification { | ||
|
|
||
| protected Type type; | ||
| protected String path; | ||
| protected long versionNum; | ||
|
|
||
| public Modification(Type type, String path, long versionNum) { | ||
| this.type = type; | ||
| this.path = path; | ||
| this.versionNum = versionNum; | ||
| } | ||
|
|
||
| public String getPath() { | ||
| return path; | ||
| } | ||
|
|
||
| public void setPath(String path) { | ||
| this.path = path; | ||
| } | ||
|
|
||
| public long getVersionNum() { | ||
| return versionNum; | ||
| } | ||
|
|
||
| public void setVersionNum(long versionNum) { | ||
| this.versionNum = versionNum; | ||
| } | ||
|
|
||
| public Type getType() { | ||
| return type; | ||
| } | ||
|
|
||
| public void setType(Type type) { | ||
| this.type = type; | ||
| } | ||
|
|
||
| public enum Type { | ||
| DELETION | ||
| } | ||
|
|
||
| @Override | ||
| public boolean equals(Object obj) { | ||
| if (!(obj instanceof Modification)) | ||
| return false; | ||
| Modification mod = (Modification) obj; | ||
| return mod.type.equals(this.type) && mod.path.equals(this.path) | ||
| && mod.versionNum == this.versionNum; | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,101 @@ | ||
| /** | ||
| * Copyright © 2019 Apache IoTDB(incubating) (dev@iotdb.apache.org) | ||
| * <p> | ||
| * Licensed 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 | ||
| * <p> | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * <p> | ||
| * 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.iotdb.db.engine.modification; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.ArrayList; | ||
| import java.util.Collection; | ||
|
|
||
| import org.apache.iotdb.db.engine.modification.io.LocalTextModificationAccessor; | ||
| import org.apache.iotdb.db.engine.modification.io.ModificationReader; | ||
| import org.apache.iotdb.db.engine.modification.io.ModificationWriter; | ||
|
|
||
| /** | ||
| * ModificationFile stores the Modifications of a TsFile or unseq file in another file in the same | ||
| * directory. | ||
| * Methods in this class are highly synchronized for concurrency safety. | ||
| */ | ||
| public class ModificationFile { | ||
|
|
||
| private Collection<Modification> modifications; | ||
| private String filePath; | ||
| private ModificationWriter writer; | ||
| private ModificationReader reader; | ||
|
|
||
| /** | ||
| * Construct a ModificationFile using a file as its storage. | ||
| * @param filePath the path of the storage file. | ||
| * @throws IOException when IOException raised when | ||
| */ | ||
| public ModificationFile(String filePath) throws IOException { | ||
| this.filePath = filePath; | ||
| LocalTextModificationAccessor accessor = new LocalTextModificationAccessor(filePath); | ||
| this.writer = accessor; | ||
| this.reader = accessor; | ||
| } | ||
|
|
||
| private void init() throws IOException { | ||
| synchronized (this) { | ||
| Collection<Modification> mods = reader.read(); | ||
| if (mods == null) { | ||
| mods = new ArrayList<>(); | ||
| } | ||
| modifications = mods; | ||
| } | ||
| } | ||
|
|
||
| private void checkInit() throws IOException { | ||
| if (modifications == null) { | ||
| init(); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Release the resources such as streams and caches. | ||
| */ | ||
| public void close() throws IOException { | ||
| synchronized (this) { | ||
| writer.close(); | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. you need to call reader.close() here. Otherwise please set
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. reader and writer refer to the same object in this implementation, so there is no need to close reader after writer is closed. |
||
| modifications = null; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Write a modification in this file. The modification will first be written to the persistent | ||
| * store then the memory cache. | ||
| * @param mod the modification to be written. | ||
| * @throws IOException if IOException is thrown when writing the modification to the store. | ||
| */ | ||
| public void write(Modification mod) throws IOException { | ||
| synchronized (this) { | ||
| checkInit(); | ||
| writer.write(mod); | ||
| modifications.add(mod); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Get all modifications stored in this file. | ||
| * @return | ||
| */ | ||
| public Collection<Modification> getModifications() throws IOException { | ||
| synchronized (this) { | ||
| checkInit(); | ||
| return new ArrayList<>(modifications); | ||
| } | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,135 @@ | ||
| /** | ||
| * Copyright © 2019 Apache IoTDB(incubating) (dev@iotdb.apache.org) | ||
| * <p> | ||
| * Licensed 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 | ||
| * <p> | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * <p> | ||
| * 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.iotdb.db.engine.modification.io; | ||
|
|
||
| import java.io.BufferedReader; | ||
| import java.io.BufferedWriter; | ||
| import java.io.FileNotFoundException; | ||
| import java.io.FileReader; | ||
| import java.io.FileWriter; | ||
| import java.io.IOException; | ||
| import java.util.ArrayList; | ||
| import java.util.Collection; | ||
| import java.util.List; | ||
|
|
||
| import org.apache.iotdb.db.engine.modification.Deletion; | ||
| import org.apache.iotdb.db.engine.modification.Modification; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| /** | ||
| * LocalTextModificationAccessor uses a file on local file system to store the modifications | ||
| * in text format, and writes modifications by appending to the tail of the file. | ||
| */ | ||
| public class LocalTextModificationAccessor implements ModificationReader, ModificationWriter { | ||
|
|
||
| private static final Logger logger = LoggerFactory.getLogger(LocalTextModificationAccessor.class); | ||
| private static final String SEPARATOR = ","; | ||
|
|
||
| private String filePath; | ||
| private BufferedWriter writer; | ||
|
|
||
| /** | ||
| * Construct a LocalTextModificationAccessor using a file specified by filePath. Only a writer | ||
| * will be created because the reader will be created only if necessary(call of read()). | ||
| * | ||
| * @param filePath the path of the file that is used for storing modifications. | ||
| * @throws IOException if the writer cannot be created. | ||
| */ | ||
| public LocalTextModificationAccessor(String filePath) throws IOException { | ||
| this.filePath = filePath; | ||
| writer = new BufferedWriter(new FileWriter(filePath)); | ||
| } | ||
|
|
||
| @Override | ||
| public Collection<Modification> read() throws IOException { | ||
| BufferedReader reader; | ||
| try { | ||
| reader = new BufferedReader(new FileReader(filePath)); | ||
| } catch (FileNotFoundException e) { | ||
| return null; | ||
| } | ||
| String line; | ||
|
|
||
| List<Modification> modificationList = new ArrayList<>(); | ||
| try { | ||
| while ((line = reader.readLine()) != null) { | ||
| modificationList.add(decodeModification(line)); | ||
| } | ||
| } catch (IOException e) { | ||
| reader.close(); | ||
| logger.error("An error occurred when reading modifications, and the remaining modifications " | ||
| + "were ignored.", e); | ||
| } | ||
| return modificationList; | ||
| } | ||
|
|
||
| @Override | ||
| public void close() throws IOException { | ||
| writer.close(); | ||
| } | ||
|
|
||
| @Override | ||
| public void write(Modification mod) throws IOException { | ||
| writer.write(encodeModification(mod)); | ||
| writer.newLine(); | ||
| writer.flush(); | ||
| } | ||
|
|
||
| private static String encodeModification(Modification mod) { | ||
| if (mod instanceof Deletion) | ||
| return encodeDeletion((Deletion) mod); | ||
| return null; | ||
| } | ||
|
|
||
| private static Modification decodeModification(String src) throws IOException { | ||
| String[] fields = src.split(SEPARATOR); | ||
| if (Modification.Type.DELETION.name().equals(fields[0])) { | ||
|
jixuan1989 marked this conversation as resolved.
|
||
| return decodeDeletion(fields); | ||
| } | ||
| throw new IOException("Unknown modification type: " + fields[0]); | ||
| } | ||
|
|
||
| private static String encodeDeletion(Deletion del) { | ||
| StringBuilder stringBuilder = new StringBuilder(); | ||
| stringBuilder.append(del.getType().toString()).append(SEPARATOR).append(del.getPath()) | ||
| .append(SEPARATOR).append(del.getVersionNum()).append(SEPARATOR) | ||
| .append(del.getTimestamp()); | ||
| return stringBuilder.toString(); | ||
| } | ||
|
|
||
| private static Deletion decodeDeletion(String[] fields) throws IOException { | ||
| if (fields.length != 4) { | ||
| throw new IOException("Incorrect deletion fields number: " + fields.length); | ||
| } | ||
|
|
||
| String path = fields[1]; | ||
| long versionNum, timestamp; | ||
| try { | ||
| versionNum = Long.parseLong(fields[2]); | ||
| } catch (NumberFormatException e) { | ||
| throw new IOException("Invalide version number: " + fields[2]); | ||
| } | ||
| try { | ||
| timestamp = Long.parseLong(fields[3]); | ||
| } catch (NumberFormatException e) { | ||
| throw new IOException("Invalide timestamp: " + fields[3]); | ||
| } | ||
|
|
||
| return new Deletion(path, versionNum, timestamp); | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,40 @@ | ||
| /** | ||
| * Copyright © 2019 Apache IoTDB(incubating) (dev@iotdb.apache.org) | ||
| * <p> | ||
| * Licensed 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 | ||
| * <p> | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * <p> | ||
| * 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.iotdb.db.engine.modification.io; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Collection; | ||
|
|
||
| import org.apache.iotdb.db.engine.modification.Modification; | ||
|
|
||
| /** | ||
| * ModificationReader reads all modifications from a persistent medium like file system. | ||
| */ | ||
| public interface ModificationReader { | ||
|
|
||
| /** | ||
| * Read all modifications from a persistent medium. | ||
| * | ||
| * @return a list of modifications contained the medium. | ||
| */ | ||
| Collection<Modification> read() throws IOException; | ||
|
|
||
| /** | ||
| * Release resources like streams. | ||
| */ | ||
| void close() throws IOException; | ||
| } |
Uh oh!
There was an error while loading. Please reload this page.