You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ch...@apache.org on 2017/01/16 14:53:18 UTC
[41/52] [partial] incubator-carbondata git commit: move core package
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/AlluxioCarbonFile.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/AlluxioCarbonFile.java b/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/AlluxioCarbonFile.java
deleted file mode 100644
index 9ff0f59..0000000
--- a/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/AlluxioCarbonFile.java
+++ /dev/null
@@ -1,130 +0,0 @@
-/*
- * 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.carbondata.core.datastorage.store.filesystem;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.carbondata.common.logging.LogService;
-import org.apache.carbondata.common.logging.LogServiceFactory;
-import org.apache.carbondata.core.datastorage.store.impl.FileFactory;
-
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hdfs.DistributedFileSystem;
-
-
-
-public class AlluxioCarbonFile extends AbstractDFSCarbonFile {
- /**
- * LOGGER
- */
- private static final LogService LOGGER =
- LogServiceFactory.getLogService(AlluxioCarbonFile.class.getName());
-
- public AlluxioCarbonFile(String filePath) {
- super(filePath);
- }
-
- public AlluxioCarbonFile(Path path) {
- super(path);
- }
-
- public AlluxioCarbonFile(FileStatus fileStatus) {
- super(fileStatus);
- }
-
- /**
- * @param listStatus
- * @return
- */
- private CarbonFile[] getFiles(FileStatus[] listStatus) {
- if (listStatus == null) {
- return new CarbonFile[0];
- }
- CarbonFile[] files = new CarbonFile[listStatus.length];
- for (int i = 0; i < files.length; i++) {
- files[i] = new AlluxioCarbonFile(listStatus[i]);
- }
- return files;
- }
-
- @Override
- public CarbonFile[] listFiles() {
- FileStatus[] listStatus = null;
- try {
- if (null != fileStatus && fileStatus.isDirectory()) {
- Path path = fileStatus.getPath();
- listStatus = path.getFileSystem(FileFactory.getConfiguration()).listStatus(path);
- } else {
- return null;
- }
- } catch (IOException e) {
- LOGGER.error("Exception occured: " + e.getMessage());
- return new CarbonFile[0];
- }
- return getFiles(listStatus);
- }
-
- @Override
- public CarbonFile[] listFiles(final CarbonFileFilter fileFilter) {
- CarbonFile[] files = listFiles();
- if (files != null && files.length >= 1) {
- List<CarbonFile> fileList = new ArrayList<CarbonFile>(files.length);
- for (int i = 0; i < files.length; i++) {
- if (fileFilter.accept(files[i])) {
- fileList.add(files[i]);
- }
- }
- if (fileList.size() >= 1) {
- return fileList.toArray(new CarbonFile[fileList.size()]);
- } else {
- return new CarbonFile[0];
- }
- }
- return files;
- }
-
- @Override
- public CarbonFile getParentFile() {
- Path parent = fileStatus.getPath().getParent();
- return null == parent ? null : new AlluxioCarbonFile(parent);
- }
-
- @Override
- public boolean renameForce(String changetoName) {
- FileSystem fs;
- try {
- fs = fileStatus.getPath().getFileSystem(FileFactory.getConfiguration());
- if (fs instanceof DistributedFileSystem) {
- ((DistributedFileSystem) fs).rename(fileStatus.getPath(), new Path(changetoName),
- org.apache.hadoop.fs.Options.Rename.OVERWRITE);
- return true;
- } else {
- return false;
- }
- } catch (IOException e) {
- LOGGER.error("Exception occured: " + e.getMessage());
- return false;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/CarbonFile.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/CarbonFile.java b/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/CarbonFile.java
deleted file mode 100644
index 642055b..0000000
--- a/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/CarbonFile.java
+++ /dev/null
@@ -1,66 +0,0 @@
-/*
- * 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.carbondata.core.datastorage.store.filesystem;
-
-public interface CarbonFile {
-
- String getAbsolutePath();
-
- CarbonFile[] listFiles(CarbonFileFilter fileFilter);
-
- CarbonFile[] listFiles();
-
- String getName();
-
- boolean isDirectory();
-
- boolean exists();
-
- String getCanonicalPath();
-
- CarbonFile getParentFile();
-
- String getPath();
-
- long getSize();
-
- boolean renameTo(String changetoName);
-
- boolean renameForce(String changetoName);
-
- boolean delete();
-
- boolean createNewFile();
-
- long getLastModifiedTime();
-
- boolean setLastModifiedTime(long timestamp);
-
- boolean truncate(String fileName, long validDataEndOffset);
-
- /**
- * This method will be used to check whether a file has been modified or not
- *
- * @param fileTimeStamp time to be compared with latest timestamp of file
- * @param endOffset file length to be compared with current length of file
- * @return
- */
- boolean isFileModified(long fileTimeStamp, long endOffset);
-}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/CarbonFileFilter.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/CarbonFileFilter.java b/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/CarbonFileFilter.java
deleted file mode 100644
index 7db3b2b..0000000
--- a/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/CarbonFileFilter.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * 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.carbondata.core.datastorage.store.filesystem;
-
-public interface CarbonFileFilter {
- boolean accept(CarbonFile file);
-}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/HDFSCarbonFile.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/HDFSCarbonFile.java b/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/HDFSCarbonFile.java
deleted file mode 100644
index ebe18e4..0000000
--- a/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/HDFSCarbonFile.java
+++ /dev/null
@@ -1,128 +0,0 @@
-/*
- * 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.carbondata.core.datastorage.store.filesystem;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.carbondata.common.logging.LogService;
-import org.apache.carbondata.common.logging.LogServiceFactory;
-import org.apache.carbondata.core.datastorage.store.impl.FileFactory;
-
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hdfs.DistributedFileSystem;
-
-public class HDFSCarbonFile extends AbstractDFSCarbonFile {
- /**
- * LOGGER
- */
- private static final LogService LOGGER =
- LogServiceFactory.getLogService(HDFSCarbonFile.class.getName());
-
- public HDFSCarbonFile(String filePath) {
- super(filePath);
- }
-
- public HDFSCarbonFile(Path path) {
- super(path);
- }
-
- public HDFSCarbonFile(FileStatus fileStatus) {
- super(fileStatus);
- }
-
- /**
- * @param listStatus
- * @return
- */
- private CarbonFile[] getFiles(FileStatus[] listStatus) {
- if (listStatus == null) {
- return new CarbonFile[0];
- }
- CarbonFile[] files = new CarbonFile[listStatus.length];
- for (int i = 0; i < files.length; i++) {
- files[i] = new HDFSCarbonFile(listStatus[i]);
- }
- return files;
- }
-
- @Override
- public CarbonFile[] listFiles() {
- FileStatus[] listStatus = null;
- try {
- if (null != fileStatus && fileStatus.isDirectory()) {
- Path path = fileStatus.getPath();
- listStatus = path.getFileSystem(FileFactory.getConfiguration()).listStatus(path);
- } else {
- return null;
- }
- } catch (IOException e) {
- LOGGER.error("Exception occured: " + e.getMessage());
- return new CarbonFile[0];
- }
- return getFiles(listStatus);
- }
-
- @Override
- public CarbonFile[] listFiles(final CarbonFileFilter fileFilter) {
- CarbonFile[] files = listFiles();
- if (files != null && files.length >= 1) {
- List<CarbonFile> fileList = new ArrayList<CarbonFile>(files.length);
- for (int i = 0; i < files.length; i++) {
- if (fileFilter.accept(files[i])) {
- fileList.add(files[i]);
- }
- }
- if (fileList.size() >= 1) {
- return fileList.toArray(new CarbonFile[fileList.size()]);
- } else {
- return new CarbonFile[0];
- }
- }
- return files;
- }
-
- @Override
- public CarbonFile getParentFile() {
- Path parent = fileStatus.getPath().getParent();
- return null == parent ? null : new HDFSCarbonFile(parent);
- }
-
- @Override
- public boolean renameForce(String changetoName) {
- FileSystem fs;
- try {
- fs = fileStatus.getPath().getFileSystem(FileFactory.getConfiguration());
- if (fs instanceof DistributedFileSystem) {
- ((DistributedFileSystem) fs).rename(fileStatus.getPath(), new Path(changetoName),
- org.apache.hadoop.fs.Options.Rename.OVERWRITE);
- return true;
- } else {
- return false;
- }
- } catch (IOException e) {
- LOGGER.error("Exception occured: " + e.getMessage());
- return false;
- }
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/LocalCarbonFile.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/LocalCarbonFile.java b/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/LocalCarbonFile.java
deleted file mode 100644
index 406f6d1..0000000
--- a/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/LocalCarbonFile.java
+++ /dev/null
@@ -1,229 +0,0 @@
-/*
- * 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.carbondata.core.datastorage.store.filesystem;
-
-import java.io.File;
-import java.io.FileFilter;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.nio.channels.FileChannel;
-
-import org.apache.carbondata.common.logging.LogService;
-import org.apache.carbondata.common.logging.LogServiceFactory;
-import org.apache.carbondata.core.constants.CarbonCommonConstants;
-import org.apache.carbondata.core.datastorage.store.impl.FileFactory;
-import org.apache.carbondata.core.util.CarbonUtil;
-
-import org.apache.hadoop.fs.Path;
-
-public class LocalCarbonFile implements CarbonFile {
- private static final LogService LOGGER =
- LogServiceFactory.getLogService(LocalCarbonFile.class.getName());
- private File file;
-
- public LocalCarbonFile(String filePath) {
- Path pathWithoutSchemeAndAuthority = Path.getPathWithoutSchemeAndAuthority(new Path(filePath));
- file = new File(pathWithoutSchemeAndAuthority.toString());
- }
-
- public LocalCarbonFile(File file) {
- this.file = file;
- }
-
- @Override public String getAbsolutePath() {
- return file.getAbsolutePath();
- }
-
- @Override public CarbonFile[] listFiles(final CarbonFileFilter fileFilter) {
- if (!file.isDirectory()) {
- return null;
- }
-
- File[] files = file.listFiles(new FileFilter() {
-
- @Override public boolean accept(File pathname) {
- return fileFilter.accept(new LocalCarbonFile(pathname));
- }
- });
-
- if (files == null) {
- return new CarbonFile[0];
- }
-
- CarbonFile[] carbonFiles = new CarbonFile[files.length];
-
- for (int i = 0; i < carbonFiles.length; i++) {
- carbonFiles[i] = new LocalCarbonFile(files[i]);
- }
-
- return carbonFiles;
- }
-
- @Override public String getName() {
- return file.getName();
- }
-
- @Override public boolean isDirectory() {
- return file.isDirectory();
- }
-
- @Override public boolean exists() {
- if (file != null) {
- return file.exists();
- }
- return false;
- }
-
- @Override public String getCanonicalPath() {
- try {
- return file.getCanonicalPath();
- } catch (IOException e) {
- LOGGER
- .error(e, "Exception occured" + e.getMessage());
- }
- return null;
- }
-
- @Override public CarbonFile getParentFile() {
- return new LocalCarbonFile(file.getParentFile());
- }
-
- @Override public String getPath() {
- return file.getPath();
- }
-
- @Override public long getSize() {
- return file.length();
- }
-
- public boolean renameTo(String changetoName) {
- return file.renameTo(new File(changetoName));
- }
-
- public boolean delete() {
- return file.delete();
- }
-
- @Override public CarbonFile[] listFiles() {
-
- if (!file.isDirectory()) {
- return null;
- }
- File[] files = file.listFiles();
- if (files == null) {
- return new CarbonFile[0];
- }
- CarbonFile[] carbonFiles = new CarbonFile[files.length];
- for (int i = 0; i < carbonFiles.length; i++) {
- carbonFiles[i] = new LocalCarbonFile(files[i]);
- }
-
- return carbonFiles;
-
- }
-
- @Override public boolean createNewFile() {
- try {
- return file.createNewFile();
- } catch (IOException e) {
- return false;
- }
- }
-
- @Override public long getLastModifiedTime() {
- return file.lastModified();
- }
-
- @Override public boolean setLastModifiedTime(long timestamp) {
- return file.setLastModified(timestamp);
- }
-
- /**
- * This method will delete the data in file data from a given offset
- */
- @Override public boolean truncate(String fileName, long validDataEndOffset) {
- FileChannel source = null;
- FileChannel destination = null;
- boolean fileTruncatedSuccessfully = false;
- // temporary file name
- String tempWriteFilePath = fileName + CarbonCommonConstants.TEMPWRITEFILEEXTENSION;
- FileFactory.FileType fileType = FileFactory.getFileType(fileName);
- try {
- CarbonFile tempFile = null;
- // delete temporary file if it already exists at a given path
- if (FileFactory.isFileExist(tempWriteFilePath, fileType)) {
- tempFile = FileFactory.getCarbonFile(tempWriteFilePath, fileType);
- tempFile.delete();
- }
- // create new temporary file
- FileFactory.createNewFile(tempWriteFilePath, fileType);
- tempFile = FileFactory.getCarbonFile(tempWriteFilePath, fileType);
- source = new FileInputStream(fileName).getChannel();
- destination = new FileOutputStream(tempWriteFilePath).getChannel();
- long read = destination.transferFrom(source, 0, validDataEndOffset);
- long totalBytesRead = read;
- long remaining = validDataEndOffset - totalBytesRead;
- // read till required data offset is not reached
- while (remaining > 0) {
- read = destination.transferFrom(source, totalBytesRead, remaining);
- totalBytesRead = totalBytesRead + read;
- remaining = remaining - totalBytesRead;
- }
- CarbonUtil.closeStreams(source, destination);
- // rename the temp file to original file
- tempFile.renameForce(fileName);
- fileTruncatedSuccessfully = true;
- } catch (IOException e) {
- LOGGER.error("Exception occured while truncating the file " + e.getMessage());
- } finally {
- CarbonUtil.closeStreams(source, destination);
- }
- return fileTruncatedSuccessfully;
- }
-
- /**
- * This method will be used to check whether a file has been modified or not
- *
- * @param fileTimeStamp time to be compared with latest timestamp of file
- * @param endOffset file length to be compared with current length of file
- * @return
- */
- @Override public boolean isFileModified(long fileTimeStamp, long endOffset) {
- boolean isFileModified = false;
- if (getLastModifiedTime() > fileTimeStamp || getSize() > endOffset) {
- isFileModified = true;
- }
- return isFileModified;
- }
-
- @Override public boolean renameForce(String changetoName) {
- File destFile = new File(changetoName);
- if (destFile.exists()) {
- if (destFile.delete()) {
- return file.renameTo(new File(changetoName));
- }
- }
-
- return file.renameTo(new File(changetoName));
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/ViewFSCarbonFile.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/ViewFSCarbonFile.java b/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/ViewFSCarbonFile.java
deleted file mode 100644
index 8f11b7a..0000000
--- a/core/src/main/java/org/apache/carbondata/core/datastorage/store/filesystem/ViewFSCarbonFile.java
+++ /dev/null
@@ -1,126 +0,0 @@
-/*
- * 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.carbondata.core.datastorage.store.filesystem;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.carbondata.common.logging.LogService;
-import org.apache.carbondata.common.logging.LogServiceFactory;
-import org.apache.carbondata.core.datastorage.store.impl.FileFactory;
-
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.viewfs.ViewFileSystem;
-
-public class ViewFSCarbonFile extends AbstractDFSCarbonFile {
- /**
- * LOGGER
- */
- private static final LogService LOGGER =
- LogServiceFactory.getLogService(ViewFSCarbonFile.class.getName());
-
- public ViewFSCarbonFile(String filePath) {
- super(filePath);
- }
-
- public ViewFSCarbonFile(Path path) {
- super(path);
- }
-
- public ViewFSCarbonFile(FileStatus fileStatus) {
- super(fileStatus);
- }
-
- /**
- * @param listStatus
- * @return
- */
- private CarbonFile[] getFiles(FileStatus[] listStatus) {
- if (listStatus == null) {
- return new CarbonFile[0];
- }
- CarbonFile[] files = new CarbonFile[listStatus.length];
- for (int i = 0; i < files.length; i++) {
- files[i] = new ViewFSCarbonFile(listStatus[i]);
- }
- return files;
- }
-
- @Override
- public CarbonFile[] listFiles() {
- FileStatus[] listStatus = null;
- try {
- if (null != fileStatus && fileStatus.isDirectory()) {
- Path path = fileStatus.getPath();
- listStatus = path.getFileSystem(FileFactory.getConfiguration()).listStatus(path);
- } else {
- return null;
- }
- } catch (IOException ex) {
- LOGGER.error("Exception occured" + ex.getMessage());
- return new CarbonFile[0];
- }
- return getFiles(listStatus);
- }
-
- @Override
- public CarbonFile[] listFiles(final CarbonFileFilter fileFilter) {
- CarbonFile[] files = listFiles();
- if (files != null && files.length >= 1) {
- List<CarbonFile> fileList = new ArrayList<CarbonFile>(files.length);
- for (int i = 0; i < files.length; i++) {
- if (fileFilter.accept(files[i])) {
- fileList.add(files[i]);
- }
- }
- if (fileList.size() >= 1) {
- return fileList.toArray(new CarbonFile[fileList.size()]);
- } else {
- return new CarbonFile[0];
- }
- }
- return files;
- }
-
- @Override public CarbonFile getParentFile() {
- Path parent = fileStatus.getPath().getParent();
- return null == parent ? null : new ViewFSCarbonFile(parent);
- }
-
- @Override
- public boolean renameForce(String changetoName) {
- FileSystem fs;
- try {
- fs = fileStatus.getPath().getFileSystem(FileFactory.getConfiguration());
- if (fs instanceof ViewFileSystem) {
- fs.delete(new Path(changetoName), true);
- fs.rename(fileStatus.getPath(), new Path(changetoName));
- return true;
- } else {
- return false;
- }
- } catch (IOException e) {
- LOGGER.error("Exception occured" + e.getMessage());
- return false;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/CompressedDataMeasureDataWrapper.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/CompressedDataMeasureDataWrapper.java b/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/CompressedDataMeasureDataWrapper.java
deleted file mode 100644
index c9571b2..0000000
--- a/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/CompressedDataMeasureDataWrapper.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * 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.carbondata.core.datastorage.store.impl;
-
-import org.apache.carbondata.core.datastorage.store.MeasureDataWrapper;
-import org.apache.carbondata.core.datastorage.store.dataholder.CarbonReadDataHolder;
-
-public class CompressedDataMeasureDataWrapper implements MeasureDataWrapper {
-
- private final CarbonReadDataHolder[] values;
-
- public CompressedDataMeasureDataWrapper(final CarbonReadDataHolder[] values) {
- this.values = values;
- }
-
- @Override public CarbonReadDataHolder[] getValues() {
- return values;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/DFSFileHolderImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/DFSFileHolderImpl.java b/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/DFSFileHolderImpl.java
deleted file mode 100644
index 6e2f5d7..0000000
--- a/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/DFSFileHolderImpl.java
+++ /dev/null
@@ -1,134 +0,0 @@
-/*
- * 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.carbondata.core.datastorage.store.impl;
-
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Map.Entry;
-
-import org.apache.carbondata.core.constants.CarbonCommonConstants;
-import org.apache.carbondata.core.datastorage.store.FileHolder;
-
-import org.apache.hadoop.fs.FSDataInputStream;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-
-
-public class DFSFileHolderImpl implements FileHolder {
- /**
- * cache to hold filename and its stream
- */
- private Map<String, FSDataInputStream> fileNameAndStreamCache;
-
- public DFSFileHolderImpl() {
- this.fileNameAndStreamCache =
- new HashMap<String, FSDataInputStream>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
- }
-
- @Override public byte[] readByteArray(String filePath, long offset, int length)
- throws IOException {
- FSDataInputStream fileChannel = updateCache(filePath);
- return read(fileChannel, length, offset);
- }
-
- /**
- * This method will be used to check whether stream is already present in
- * cache or not for filepath if not present then create it and then add to
- * cache, other wise get from cache
- *
- * @param filePath fully qualified file path
- * @return channel
- */
- private FSDataInputStream updateCache(String filePath) throws IOException {
- FSDataInputStream fileChannel = fileNameAndStreamCache.get(filePath);
- if (null == fileChannel) {
- Path pt = new Path(filePath);
- FileSystem fs = FileSystem.get(FileFactory.getConfiguration());
- fileChannel = fs.open(pt);
- fileNameAndStreamCache.put(filePath, fileChannel);
- }
- return fileChannel;
- }
-
- /**
- * This method will be used to read from file based on number of bytes to be read and positon
- *
- * @param channel file channel
- * @param size number of bytes
- * @param offset position
- * @return byte buffer
- */
- private byte[] read(FSDataInputStream channel, int size, long offset) throws IOException {
- byte[] byteBffer = new byte[size];
- channel.seek(offset);
- channel.readFully(byteBffer);
- return byteBffer;
- }
-
- /**
- * This method will be used to read from file based on number of bytes to be read and positon
- *
- * @param channel file channel
- * @param size number of bytes
- * @return byte buffer
- */
- private byte[] read(FSDataInputStream channel, int size) throws IOException {
- byte[] byteBffer = new byte[size];
- channel.readFully(byteBffer);
- return byteBffer;
- }
-
- @Override public int readInt(String filePath, long offset) throws IOException {
- FSDataInputStream fileChannel = updateCache(filePath);
- fileChannel.seek(offset);
- return fileChannel.readInt();
- }
-
- @Override public long readDouble(String filePath, long offset) throws IOException {
- FSDataInputStream fileChannel = updateCache(filePath);
- fileChannel.seek(offset);
- return fileChannel.readLong();
- }
-
- @Override public void finish() throws IOException {
- for (Entry<String, FSDataInputStream> entry : fileNameAndStreamCache.entrySet()) {
- FSDataInputStream channel = entry.getValue();
- if (null != channel) {
- channel.close();
- }
- }
- }
-
- @Override public byte[] readByteArray(String filePath, int length) throws IOException {
- FSDataInputStream fileChannel = updateCache(filePath);
- return read(fileChannel, length);
- }
-
- @Override public long readLong(String filePath, long offset) throws IOException {
- FSDataInputStream fileChannel = updateCache(filePath);
- fileChannel.seek(offset);
- return fileChannel.readLong();
- }
-
- @Override public int readInt(String filePath) throws IOException {
- FSDataInputStream fileChannel = updateCache(filePath);
- return fileChannel.readInt();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/FileFactory.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/FileFactory.java b/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/FileFactory.java
deleted file mode 100644
index f0c424b..0000000
--- a/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/FileFactory.java
+++ /dev/null
@@ -1,485 +0,0 @@
-/*
- * 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.carbondata.core.datastorage.store.impl;
-
-import java.io.BufferedInputStream;
-import java.io.BufferedOutputStream;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.zip.GZIPInputStream;
-
-import org.apache.carbondata.core.datastorage.store.FileHolder;
-import org.apache.carbondata.core.datastorage.store.filesystem.AlluxioCarbonFile;
-import org.apache.carbondata.core.datastorage.store.filesystem.CarbonFile;
-import org.apache.carbondata.core.datastorage.store.filesystem.HDFSCarbonFile;
-import org.apache.carbondata.core.datastorage.store.filesystem.LocalCarbonFile;
-import org.apache.carbondata.core.datastorage.store.filesystem.ViewFSCarbonFile;
-import org.apache.carbondata.core.util.CarbonUtil;
-
-import org.apache.commons.compress.compressors.bzip2.BZip2CompressorInputStream;
-import org.apache.commons.io.FileUtils;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FSDataInputStream;
-import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.compress.BZip2Codec;
-import org.apache.hadoop.io.compress.GzipCodec;
-
-public final class FileFactory {
- private static Configuration configuration = null;
-
- static {
- configuration = new Configuration();
- configuration.addResource(new Path("../core-default.xml"));
- }
-
- private FileFactory() {
-
- }
-
- public static Configuration getConfiguration() {
- return configuration;
- }
-
- public static FileHolder getFileHolder(FileType fileType) {
- switch (fileType) {
- case LOCAL:
- return new FileHolderImpl();
- case HDFS:
- case ALLUXIO:
- case VIEWFS:
- return new DFSFileHolderImpl();
- default:
- return new FileHolderImpl();
- }
- }
-
- public static FileType getFileType(String path) {
- if (path.startsWith(CarbonUtil.HDFS_PREFIX)) {
- return FileType.HDFS;
- }
- else if (path.startsWith(CarbonUtil.ALLUXIO_PREFIX)) {
- return FileType.ALLUXIO;
- }
- else if (path.startsWith(CarbonUtil.VIEWFS_PREFIX)) {
- return FileType.VIEWFS;
- }
- return FileType.LOCAL;
- }
-
- public static CarbonFile getCarbonFile(String path, FileType fileType) {
- switch (fileType) {
- case LOCAL:
- return new LocalCarbonFile(path);
- case HDFS:
- return new HDFSCarbonFile(path);
- case ALLUXIO:
- return new AlluxioCarbonFile(path);
- case VIEWFS:
- return new ViewFSCarbonFile(path);
- default:
- return new LocalCarbonFile(path);
- }
- }
-
- public static DataInputStream getDataInputStream(String path, FileType fileType)
- throws IOException {
- return getDataInputStream(path, fileType, -1);
- }
-
- public static DataInputStream getDataInputStream(String path, FileType fileType, int bufferSize)
- throws IOException {
- path = path.replace("\\", "/");
- boolean gzip = path.endsWith(".gz");
- boolean bzip2 = path.endsWith(".bz2");
- InputStream stream;
- switch (fileType) {
- case LOCAL:
- if (gzip) {
- stream = new GZIPInputStream(new FileInputStream(path));
- } else if (bzip2) {
- stream = new BZip2CompressorInputStream(new FileInputStream(path));
- } else {
- stream = new FileInputStream(path);
- }
- break;
- case HDFS:
- case ALLUXIO:
- case VIEWFS:
- Path pt = new Path(path);
- FileSystem fs = pt.getFileSystem(configuration);
- if (bufferSize == -1) {
- stream = fs.open(pt);
- } else {
- stream = fs.open(pt, bufferSize);
- }
- if (gzip) {
- GzipCodec codec = new GzipCodec();
- stream = codec.createInputStream(stream);
- } else if (bzip2) {
- BZip2Codec codec = new BZip2Codec();
- stream = codec.createInputStream(stream);
- }
- break;
- default:
- throw new UnsupportedOperationException("unsupported file system");
- }
- return new DataInputStream(new BufferedInputStream(stream));
- }
-
- /**
- * return the datainputStream which is seek to the offset of file
- *
- * @param path
- * @param fileType
- * @param bufferSize
- * @param offset
- * @return DataInputStream
- * @throws IOException
- */
- public static DataInputStream getDataInputStream(String path, FileType fileType, int bufferSize,
- long offset) throws IOException {
- path = path.replace("\\", "/");
- switch (fileType) {
- case HDFS:
- case ALLUXIO:
- case VIEWFS:
- Path pt = new Path(path);
- FileSystem fs = pt.getFileSystem(configuration);
- FSDataInputStream stream = fs.open(pt, bufferSize);
- stream.seek(offset);
- return new DataInputStream(new BufferedInputStream(stream));
- default:
- FileInputStream fis = new FileInputStream(path);
- long actualSkipSize = 0;
- long skipSize = offset;
- while (actualSkipSize != offset) {
- actualSkipSize += fis.skip(skipSize);
- skipSize = skipSize - actualSkipSize;
- }
- return new DataInputStream(new BufferedInputStream(fis));
- }
- }
-
- public static DataOutputStream getDataOutputStream(String path, FileType fileType)
- throws IOException {
- path = path.replace("\\", "/");
- switch (fileType) {
- case LOCAL:
- return new DataOutputStream(new BufferedOutputStream(new FileOutputStream(path)));
- case HDFS:
- case ALLUXIO:
- case VIEWFS:
- Path pt = new Path(path);
- FileSystem fs = pt.getFileSystem(configuration);
- FSDataOutputStream stream = fs.create(pt, true);
- return stream;
- default:
- return new DataOutputStream(new BufferedOutputStream(new FileOutputStream(path)));
- }
- }
-
- public static DataOutputStream getDataOutputStream(String path, FileType fileType, int bufferSize,
- boolean append) throws IOException {
- path = path.replace("\\", "/");
- switch (fileType) {
- case LOCAL:
- return new DataOutputStream(
- new BufferedOutputStream(new FileOutputStream(path, append), bufferSize));
- case HDFS:
- case ALLUXIO:
- case VIEWFS:
- Path pt = new Path(path);
- FileSystem fs = pt.getFileSystem(configuration);
- FSDataOutputStream stream = null;
- if (append) {
- // append to a file only if file already exists else file not found
- // exception will be thrown by hdfs
- if (CarbonUtil.isFileExists(path)) {
- stream = fs.append(pt, bufferSize);
- } else {
- stream = fs.create(pt, true, bufferSize);
- }
- } else {
- stream = fs.create(pt, true, bufferSize);
- }
- return stream;
- default:
- return new DataOutputStream(
- new BufferedOutputStream(new FileOutputStream(path), bufferSize));
- }
- }
-
- public static DataOutputStream getDataOutputStream(String path, FileType fileType, int bufferSize,
- long blockSize) throws IOException {
- path = path.replace("\\", "/");
- switch (fileType) {
- case LOCAL:
- return new DataOutputStream(
- new BufferedOutputStream(new FileOutputStream(path), bufferSize));
- case HDFS:
- case ALLUXIO:
- case VIEWFS:
- Path pt = new Path(path);
- FileSystem fs = pt.getFileSystem(configuration);
- FSDataOutputStream stream =
- fs.create(pt, true, bufferSize, fs.getDefaultReplication(pt), blockSize);
- return stream;
- default:
- return new DataOutputStream(
- new BufferedOutputStream(new FileOutputStream(path), bufferSize));
- }
- }
-
- /**
- * This method checks the given path exists or not and also is it file or
- * not if the performFileCheck is true
- *
- * @param filePath - Path
- * @param fileType - FileType Local/HDFS
- * @param performFileCheck - Provide false for folders, true for files and
- */
- public static boolean isFileExist(String filePath, FileType fileType, boolean performFileCheck)
- throws IOException {
- filePath = filePath.replace("\\", "/");
- switch (fileType) {
- case HDFS:
- case ALLUXIO:
- case VIEWFS:
- Path path = new Path(filePath);
- FileSystem fs = path.getFileSystem(configuration);
- if (performFileCheck) {
- return fs.exists(path) && fs.isFile(path);
- } else {
- return fs.exists(path);
- }
-
- case LOCAL:
- default:
- File defaultFile = new File(filePath);
-
- if (performFileCheck) {
- return defaultFile.exists() && defaultFile.isFile();
- } else {
- return defaultFile.exists();
- }
- }
- }
-
- /**
- * This method checks the given path exists or not and also is it file or
- * not if the performFileCheck is true
- *
- * @param filePath - Path
- * @param fileType - FileType Local/HDFS
- */
- public static boolean isFileExist(String filePath, FileType fileType) throws IOException {
- filePath = filePath.replace("\\", "/");
- switch (fileType) {
- case HDFS:
- case ALLUXIO:
- case VIEWFS:
- Path path = new Path(filePath);
- FileSystem fs = path.getFileSystem(configuration);
- return fs.exists(path);
-
- case LOCAL:
- default:
- File defaultFile = new File(filePath);
- return defaultFile.exists();
- }
- }
-
- public static boolean createNewFile(String filePath, FileType fileType) throws IOException {
- filePath = filePath.replace("\\", "/");
- switch (fileType) {
- case HDFS:
- case ALLUXIO:
- case VIEWFS:
- Path path = new Path(filePath);
- FileSystem fs = path.getFileSystem(configuration);
- return fs.createNewFile(path);
-
- case LOCAL:
- default:
- File file = new File(filePath);
- return file.createNewFile();
- }
- }
-
- public static boolean deleteFile(String filePath, FileType fileType) throws IOException {
- filePath = filePath.replace("\\", "/");
- switch (fileType) {
- case HDFS:
- case ALLUXIO:
- case VIEWFS:
- Path path = new Path(filePath);
- FileSystem fs = path.getFileSystem(configuration);
- return fs.delete(path, true);
-
- case LOCAL:
- default:
- File file = new File(filePath);
- return deleteAllFilesOfDir(file);
- }
- }
-
- public static boolean deleteAllFilesOfDir(File path) {
- if (!path.exists()) {
- return true;
- }
- if (path.isFile()) {
- return path.delete();
- }
- File[] files = path.listFiles();
- for (int i = 0; i < files.length; i++) {
- deleteAllFilesOfDir(files[i]);
- }
- return path.delete();
- }
-
- public static boolean mkdirs(String filePath, FileType fileType) throws IOException {
- filePath = filePath.replace("\\", "/");
- switch (fileType) {
- case HDFS:
- case ALLUXIO:
- case VIEWFS:
- Path path = new Path(filePath);
- FileSystem fs = path.getFileSystem(configuration);
- return fs.mkdirs(path);
- case LOCAL:
- default:
- File file = new File(filePath);
- return file.mkdirs();
- }
- }
-
- /**
- * for getting the dataoutput stream using the hdfs filesystem append API.
- *
- * @param path
- * @param fileType
- * @return
- * @throws IOException
- */
- public static DataOutputStream getDataOutputStreamUsingAppend(String path, FileType fileType)
- throws IOException {
- path = path.replace("\\", "/");
- switch (fileType) {
- case LOCAL:
- return new DataOutputStream(new BufferedOutputStream(new FileOutputStream(path, true)));
- case HDFS:
- case ALLUXIO:
- case VIEWFS:
- Path pt = new Path(path);
- FileSystem fs = pt.getFileSystem(configuration);
- FSDataOutputStream stream = fs.append(pt);
- return stream;
- default:
- return new DataOutputStream(new BufferedOutputStream(new FileOutputStream(path)));
- }
- }
-
- /**
- * for creating a new Lock file and if it is successfully created
- * then in case of abrupt shutdown then the stream to that file will be closed.
- *
- * @param filePath
- * @param fileType
- * @return
- * @throws IOException
- */
- public static boolean createNewLockFile(String filePath, FileType fileType) throws IOException {
- filePath = filePath.replace("\\", "/");
- switch (fileType) {
- case HDFS:
- case ALLUXIO:
- case VIEWFS:
- Path path = new Path(filePath);
- FileSystem fs = path.getFileSystem(configuration);
- if (fs.createNewFile(path)) {
- fs.deleteOnExit(path);
- return true;
- }
- return false;
- case LOCAL:
- default:
- File file = new File(filePath);
- return file.createNewFile();
- }
- }
-
- public enum FileType {
- LOCAL, HDFS, ALLUXIO, VIEWFS
- }
-
- /**
- * below method will be used to update the file path
- * for local type
- * it removes the file:/ from the path
- *
- * @param filePath
- * @return updated file path without url for local
- */
- public static String getUpdatedFilePath(String filePath) {
- FileType fileType = getFileType(filePath);
- switch (fileType) {
- case HDFS:
- case ALLUXIO:
- case VIEWFS:
- return filePath;
- case LOCAL:
- default:
- Path pathWithoutSchemeAndAuthority =
- Path.getPathWithoutSchemeAndAuthority(new Path(filePath));
- return pathWithoutSchemeAndAuthority.toString();
- }
- }
-
- /**
- * It computes size of directory
- *
- * @param filePath
- * @return size in bytes
- * @throws IOException
- */
- public static long getDirectorySize(String filePath) throws IOException {
- FileType fileType = getFileType(filePath);
- switch (fileType) {
- case HDFS:
- case ALLUXIO:
- case VIEWFS:
- Path path = new Path(filePath);
- FileSystem fs = path.getFileSystem(configuration);
- return fs.getContentSummary(path).getLength();
- case LOCAL:
- default:
- File file = new File(filePath);
- return FileUtils.sizeOfDirectory(file);
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/FileHolderImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/FileHolderImpl.java b/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/FileHolderImpl.java
deleted file mode 100644
index 9a5c06a..0000000
--- a/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/FileHolderImpl.java
+++ /dev/null
@@ -1,198 +0,0 @@
-/*
- * 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.carbondata.core.datastorage.store.impl;
-
-import java.io.FileInputStream;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.nio.channels.FileChannel;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Map.Entry;
-
-import org.apache.carbondata.core.constants.CarbonCommonConstants;
-import org.apache.carbondata.core.datastorage.store.FileHolder;
-
-public class FileHolderImpl implements FileHolder {
- /**
- * cache to hold filename and its stream
- */
- private Map<String, FileChannel> fileNameAndStreamCache;
-
- /**
- * FileHolderImpl Constructor
- * It will create the cache
- */
- public FileHolderImpl() {
- this.fileNameAndStreamCache =
- new HashMap<String, FileChannel>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
- }
-
- public FileHolderImpl(int capacity) {
- this.fileNameAndStreamCache = new HashMap<String, FileChannel>(capacity);
- }
-
- /**
- * This method will be used to read the byte array from file based on offset
- * and length(number of bytes) need to read
- *
- * @param filePath fully qualified file path
- * @param offset reading start position,
- * @param length number of bytes to be read
- * @return read byte array
- */
- @Override public byte[] readByteArray(String filePath, long offset, int length)
- throws IOException {
- FileChannel fileChannel = updateCache(filePath);
- ByteBuffer byteBffer = read(fileChannel, length, offset);
- return byteBffer.array();
- }
-
- /**
- * This method will be used to close all the streams currently present in the cache
- */
- @Override public void finish() throws IOException {
- for (Entry<String, FileChannel> entry : fileNameAndStreamCache.entrySet()) {
- FileChannel channel = entry.getValue();
- if (null != channel) {
- channel.close();
- }
- }
- }
-
- /**
- * This method will be used to read int from file from postion(offset), here
- * length will be always 4 bacause int byte size if 4
- *
- * @param filePath fully qualified file path
- * @param offset reading start position,
- * @return read int
- */
- @Override public int readInt(String filePath, long offset) throws IOException {
- FileChannel fileChannel = updateCache(filePath);
- ByteBuffer byteBffer = read(fileChannel, CarbonCommonConstants.INT_SIZE_IN_BYTE, offset);
- return byteBffer.getInt();
- }
-
- /**
- * This method will be used to read int from file from postion(offset), here
- * length will be always 4 bacause int byte size if 4
- *
- * @param filePath fully qualified file path
- * @return read int
- */
- @Override public int readInt(String filePath) throws IOException {
- FileChannel fileChannel = updateCache(filePath);
- ByteBuffer byteBffer = read(fileChannel, CarbonCommonConstants.INT_SIZE_IN_BYTE);
- return byteBffer.getInt();
- }
-
- /**
- * This method will be used to read int from file from postion(offset), here
- * length will be always 4 bacause int byte size if 4
- *
- * @param filePath fully qualified file path
- * @param offset reading start position,
- * @return read int
- */
- @Override public long readDouble(String filePath, long offset) throws IOException {
- FileChannel fileChannel = updateCache(filePath);
- ByteBuffer byteBffer = read(fileChannel, CarbonCommonConstants.LONG_SIZE_IN_BYTE, offset);
- return byteBffer.getLong();
- }
-
- /**
- * This method will be used to check whether stream is already present in
- * cache or not for filepath if not present then create it and then add to
- * cache, other wise get from cache
- *
- * @param filePath fully qualified file path
- * @return channel
- */
- private FileChannel updateCache(String filePath) throws FileNotFoundException {
- FileChannel fileChannel = fileNameAndStreamCache.get(filePath);
- if (null == fileChannel) {
- FileInputStream stream = new FileInputStream(filePath);
- fileChannel = stream.getChannel();
- fileNameAndStreamCache.put(filePath, fileChannel);
- }
- return fileChannel;
- }
-
- /**
- * This method will be used to read from file based on number of bytes to be read and positon
- *
- * @param channel file channel
- * @param size number of bytes
- * @param offset position
- * @return byte buffer
- */
- private ByteBuffer read(FileChannel channel, int size, long offset) throws IOException {
- ByteBuffer byteBffer = ByteBuffer.allocate(size);
- channel.position(offset);
- channel.read(byteBffer);
- byteBffer.rewind();
- return byteBffer;
- }
-
- /**
- * This method will be used to read from file based on number of bytes to be read and positon
- *
- * @param channel file channel
- * @param size number of bytes
- * @return byte buffer
- */
- private ByteBuffer read(FileChannel channel, int size) throws IOException {
- ByteBuffer byteBffer = ByteBuffer.allocate(size);
- channel.read(byteBffer);
- byteBffer.rewind();
- return byteBffer;
- }
-
-
- /**
- * This method will be used to read the byte array from file based on length(number of bytes)
- *
- * @param filePath fully qualified file path
- * @param length number of bytes to be read
- * @return read byte array
- */
- @Override public byte[] readByteArray(String filePath, int length) throws IOException {
- FileChannel fileChannel = updateCache(filePath);
- ByteBuffer byteBffer = read(fileChannel, length);
- return byteBffer.array();
- }
-
- /**
- * This method will be used to read long from file from postion(offset), here
- * length will be always 8 bacause int byte size is 8
- *
- * @param filePath fully qualified file path
- * @param offset reading start position,
- * @return read long
- */
- @Override public long readLong(String filePath, long offset) throws IOException {
- FileChannel fileChannel = updateCache(filePath);
- ByteBuffer byteBffer = read(fileChannel, CarbonCommonConstants.LONG_SIZE_IN_BYTE, offset);
- return byteBffer.getLong();
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/data/compressed/AbstractHeavyCompressedDoubleArrayDataStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/data/compressed/AbstractHeavyCompressedDoubleArrayDataStore.java b/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/data/compressed/AbstractHeavyCompressedDoubleArrayDataStore.java
deleted file mode 100644
index ad1d62f..0000000
--- a/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/data/compressed/AbstractHeavyCompressedDoubleArrayDataStore.java
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- * 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.carbondata.core.datastorage.store.impl.data.compressed;
-
-import org.apache.carbondata.common.logging.LogService;
-import org.apache.carbondata.common.logging.LogServiceFactory;
-import org.apache.carbondata.core.constants.CarbonCommonConstants;
-import org.apache.carbondata.core.datastorage.store.NodeMeasureDataStore;
-import org.apache.carbondata.core.datastorage.store.compression.ValueCompressionHolder;
-import org.apache.carbondata.core.datastorage.store.compression.WriterCompressModel;
-import org.apache.carbondata.core.datastorage.store.dataholder.CarbonWriteDataHolder;
-import org.apache.carbondata.core.util.ValueCompressionUtil;
-
-public abstract class AbstractHeavyCompressedDoubleArrayDataStore
- implements NodeMeasureDataStore //NodeMeasureDataStore<double[]>
-{
-
- private LogService LOGGER =
- LogServiceFactory.getLogService(AbstractHeavyCompressedDoubleArrayDataStore.class.getName());
-
- /**
- * values.
- */
- protected ValueCompressionHolder[] values;
-
- /**
- * compressionModel.
- */
- protected WriterCompressModel compressionModel;
-
- /**
- * type
- */
- private char[] type;
-
- /**
- * AbstractHeavyCompressedDoubleArrayDataStore constructor.
- *
- * @param compressionModel
- */
- public AbstractHeavyCompressedDoubleArrayDataStore(WriterCompressModel compressionModel) {
- this.compressionModel = compressionModel;
- if (null != compressionModel) {
- this.type = compressionModel.getType();
- values =
- new ValueCompressionHolder[compressionModel.getValueCompressionHolder().length];
- }
- }
-
- // this method first invokes encoding routine to encode the data chunk,
- // followed by invoking compression routine for preparing the data chunk for writing.
- @Override public byte[][] getWritableMeasureDataArray(CarbonWriteDataHolder[] dataHolder) {
- byte[][] returnValue = new byte[values.length][];
- for (int i = 0; i < compressionModel.getValueCompressionHolder().length; i++) {
- values[i] = compressionModel.getValueCompressionHolder()[i];
- if (type[i] != CarbonCommonConstants.BYTE_VALUE_MEASURE) {
- // first perform encoding of the data chunk
- values[i].setValue(
- ValueCompressionUtil.getValueCompressor(compressionModel.getCompressionFinders()[i])
- .getCompressedValues(compressionModel.getCompressionFinders()[i], dataHolder[i],
- compressionModel.getMaxValue()[i],
- compressionModel.getMantissa()[i]));
- } else {
- values[i].setValue(dataHolder[i].getWritableByteArrayValues());
- }
- values[i].compress();
- returnValue[i] = values[i].getCompressedData();
- }
-
- return returnValue;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/data/compressed/HeavyCompressedDoubleArrayDataInMemoryStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/data/compressed/HeavyCompressedDoubleArrayDataInMemoryStore.java b/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/data/compressed/HeavyCompressedDoubleArrayDataInMemoryStore.java
deleted file mode 100644
index 0c29143..0000000
--- a/core/src/main/java/org/apache/carbondata/core/datastorage/store/impl/data/compressed/HeavyCompressedDoubleArrayDataInMemoryStore.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * 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.carbondata.core.datastorage.store.impl.data.compressed;
-
-import org.apache.carbondata.core.datastorage.store.compression.WriterCompressModel;
-
-public class HeavyCompressedDoubleArrayDataInMemoryStore
- extends AbstractHeavyCompressedDoubleArrayDataStore {
-
- public HeavyCompressedDoubleArrayDataInMemoryStore(WriterCompressModel compressionModel) {
- super(compressionModel);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/datastorage/util/StoreFactory.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastorage/util/StoreFactory.java b/core/src/main/java/org/apache/carbondata/core/datastorage/util/StoreFactory.java
deleted file mode 100644
index 7561ec7..0000000
--- a/core/src/main/java/org/apache/carbondata/core/datastorage/util/StoreFactory.java
+++ /dev/null
@@ -1,35 +0,0 @@
-/*
- * 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.carbondata.core.datastorage.util;
-
-import org.apache.carbondata.core.datastorage.store.NodeMeasureDataStore;
-import org.apache.carbondata.core.datastorage.store.compression.WriterCompressModel;
-import org.apache.carbondata.core.datastorage.store.impl.data.compressed.HeavyCompressedDoubleArrayDataInMemoryStore;
-
-public final class StoreFactory {
-
- private StoreFactory() {
- }
-
- public static NodeMeasureDataStore createDataStore(WriterCompressModel compressionModel) {
- return new HeavyCompressedDoubleArrayDataInMemoryStore(compressionModel);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/datastore/AbstractBlockIndexStoreCache.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/AbstractBlockIndexStoreCache.java b/core/src/main/java/org/apache/carbondata/core/datastore/AbstractBlockIndexStoreCache.java
new file mode 100644
index 0000000..cf7bf21
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/AbstractBlockIndexStoreCache.java
@@ -0,0 +1,115 @@
+/*
+ * 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.carbondata.core.datastore;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.carbondata.core.cache.Cache;
+import org.apache.carbondata.core.cache.CarbonLRUCache;
+import org.apache.carbondata.core.datastore.block.AbstractIndex;
+import org.apache.carbondata.core.datastore.block.BlockInfo;
+import org.apache.carbondata.core.datastore.block.TableBlockInfo;
+import org.apache.carbondata.core.datastore.block.TableBlockUniqueIdentifier;
+import org.apache.carbondata.core.datastore.exception.IndexBuilderException;
+import org.apache.carbondata.core.metadata.blocklet.DataFileFooter;
+import org.apache.carbondata.core.util.CarbonUtil;
+
+/**
+ * This class validate and load the B-Tree in the executor lru cache
+ * @param <K> cache key
+ * @param <V> Block Meta data details
+ */
+public abstract class AbstractBlockIndexStoreCache<K, V>
+ implements Cache<TableBlockUniqueIdentifier, AbstractIndex> {
+ /**
+ * carbon store path
+ */
+ protected String carbonStorePath;
+ /**
+ * CarbonLRU cache
+ */
+ protected CarbonLRUCache lruCache;
+
+ /**
+ * table segment id vs blockInfo list
+ */
+ protected Map<String, List<BlockInfo>> segmentIdToBlockListMap;
+
+
+ /**
+ * map of block info to lock object map, while loading the btree this will be filled
+ * and removed after loading the tree for that particular block info, this will be useful
+ * while loading the tree concurrently so only block level lock will be applied another
+ * block can be loaded concurrently
+ */
+ protected Map<BlockInfo, Object> blockInfoLock;
+
+ /**
+ * The object will hold the segment ID lock so that at a time only 1 block that belongs to same
+ * segment & table can create the list for holding the block info
+ */
+ protected Map<String, Object> segmentIDLock;
+
+ public AbstractBlockIndexStoreCache(String carbonStorePath, CarbonLRUCache lruCache) {
+ this.carbonStorePath = carbonStorePath;
+ this.lruCache = lruCache;
+ blockInfoLock = new ConcurrentHashMap<BlockInfo, Object>();
+ segmentIDLock= new ConcurrentHashMap<String, Object>();
+ segmentIdToBlockListMap = new ConcurrentHashMap<>();
+ }
+
+ /**
+ * This method will get the value for the given key. If value does not exist
+ * for the given key, it will check and load the value.
+ *
+ * @param tableBlock
+ * @param tableBlockUniqueIdentifier
+ * @param lruCacheKey
+ */
+ protected void checkAndLoadTableBlocks(AbstractIndex tableBlock,
+ TableBlockUniqueIdentifier tableBlockUniqueIdentifier, String lruCacheKey)
+ throws IOException {
+ // calculate the required size is
+ TableBlockInfo blockInfo = tableBlockUniqueIdentifier.getTableBlockInfo();
+ long requiredMetaSize = CarbonUtil.calculateMetaSize(blockInfo);
+ if (requiredMetaSize > 0) {
+ tableBlock.setMemorySize(requiredMetaSize);
+ tableBlock.incrementAccessCount();
+ boolean isTableBlockAddedToLruCache = lruCache.put(lruCacheKey, tableBlock, requiredMetaSize);
+ // if column is successfully added to lru cache then only load the
+ // table blocks data
+ if (isTableBlockAddedToLruCache) {
+ // load table blocks data
+ // getting the data file meta data of the block
+ DataFileFooter footer = CarbonUtil.readMetadatFile(blockInfo);
+ footer.setBlockInfo(new BlockInfo(blockInfo));
+ // building the block
+ tableBlock.buildIndex(Collections.singletonList(footer));
+ } else {
+ throw new IndexBuilderException(
+ "Cannot load table blocks into memory. Not enough memory available");
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/datastore/BTreeBuilderInfo.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/BTreeBuilderInfo.java b/core/src/main/java/org/apache/carbondata/core/datastore/BTreeBuilderInfo.java
new file mode 100644
index 0000000..f1b0bf7
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/BTreeBuilderInfo.java
@@ -0,0 +1,61 @@
+/*
+ * 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.carbondata.core.datastore;
+
+import java.util.List;
+
+import org.apache.carbondata.core.metadata.blocklet.DataFileFooter;
+
+/**
+ * below class holds the meta data requires to build the blocks
+ */
+public class BTreeBuilderInfo {
+
+ /**
+ * holds all the information about data
+ * file meta data
+ */
+ private List<DataFileFooter> footerList;
+
+ /**
+ * size of the each column value size
+ * this will be useful for reading
+ */
+ private int[] dimensionColumnValueSize;
+
+ public BTreeBuilderInfo(List<DataFileFooter> footerList,
+ int[] dimensionColumnValueSize) {
+ this.dimensionColumnValueSize = dimensionColumnValueSize;
+ this.footerList = footerList;
+ }
+
+ /**
+ * @return the eachDimensionBlockSize
+ */
+ public int[] getDimensionColumnValueSize() {
+ return dimensionColumnValueSize;
+ }
+
+ /**
+ * @return the footerList
+ */
+ public List<DataFileFooter> getFooterList() {
+ return footerList;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/ce09aaaf/core/src/main/java/org/apache/carbondata/core/datastore/BlockIndexStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/BlockIndexStore.java b/core/src/main/java/org/apache/carbondata/core/datastore/BlockIndexStore.java
new file mode 100644
index 0000000..8e93a5b
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/BlockIndexStore.java
@@ -0,0 +1,380 @@
+/*
+ * 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.carbondata.core.datastore;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.cache.CarbonLRUCache;
+import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.datastore.block.AbstractIndex;
+import org.apache.carbondata.core.datastore.block.BlockIndex;
+import org.apache.carbondata.core.datastore.block.BlockInfo;
+import org.apache.carbondata.core.datastore.block.SegmentTaskIndexWrapper;
+import org.apache.carbondata.core.datastore.block.TableBlockInfo;
+import org.apache.carbondata.core.datastore.block.TableBlockUniqueIdentifier;
+import org.apache.carbondata.core.datastore.exception.IndexBuilderException;
+import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
+import org.apache.carbondata.core.metadata.CarbonTableIdentifier;
+import org.apache.carbondata.core.mutate.CarbonUpdateUtil;
+import org.apache.carbondata.core.mutate.UpdateVO;
+import org.apache.carbondata.core.scan.model.QueryModel;
+import org.apache.carbondata.core.util.CarbonProperties;
+
+/**
+ * This class is used to load the B-Tree in Executor LRU Cache
+ */
+public class BlockIndexStore<K, V> extends AbstractBlockIndexStoreCache<K, V> {
+
+ /**
+ * LOGGER instance
+ */
+ private static final LogService LOGGER =
+ LogServiceFactory.getLogService(BlockIndexStore.class.getName());
+ public BlockIndexStore(String carbonStorePath, CarbonLRUCache lruCache) {
+ super(carbonStorePath, lruCache);
+ }
+
+ /**
+ * The method loads the block meta in B-tree lru cache and returns the block meta.
+ *
+ * @param tableBlockUniqueIdentifier Uniquely identifies the block
+ * @return returns the blocks B-Tree meta
+ */
+ @Override public AbstractIndex get(TableBlockUniqueIdentifier tableBlockUniqueIdentifier)
+ throws IOException {
+ TableBlockInfo tableBlockInfo = tableBlockUniqueIdentifier.getTableBlockInfo();
+ BlockInfo blockInfo = new BlockInfo(tableBlockInfo);
+ String lruCacheKey =
+ getLruCacheKey(tableBlockUniqueIdentifier.getAbsoluteTableIdentifier(), blockInfo);
+ AbstractIndex tableBlock = (AbstractIndex) lruCache.get(lruCacheKey);
+
+ // if block is not loaded
+ if (null == tableBlock) {
+ // check any lock object is present in
+ // block info lock map
+ Object blockInfoLockObject = blockInfoLock.get(blockInfo);
+ // if lock object is not present then acquire
+ // the lock in block info lock and add a lock object in the map for
+ // particular block info, added double checking mechanism to add the lock
+ // object so in case of concurrent query we for same block info only one lock
+ // object will be added
+ if (null == blockInfoLockObject) {
+ synchronized (blockInfoLock) {
+ // again checking the block info lock, to check whether lock object is present
+ // or not if now also not present then add a lock object
+ blockInfoLockObject = blockInfoLock.get(blockInfo);
+ if (null == blockInfoLockObject) {
+ blockInfoLockObject = new Object();
+ blockInfoLock.put(blockInfo, blockInfoLockObject);
+ }
+ }
+ }
+ //acquire the lock for particular block info
+ synchronized (blockInfoLockObject) {
+ // check again whether block is present or not to avoid the
+ // same block is loaded
+ //more than once in case of concurrent query
+ tableBlock = (AbstractIndex) lruCache.get(
+ getLruCacheKey(tableBlockUniqueIdentifier.getAbsoluteTableIdentifier(), blockInfo));
+ // if still block is not present then load the block
+ if (null == tableBlock) {
+ tableBlock = loadBlock(tableBlockUniqueIdentifier);
+ fillSegmentIdToBlockListMap(tableBlockUniqueIdentifier.getAbsoluteTableIdentifier(),
+ blockInfo);
+ }
+ }
+ } else {
+ tableBlock.incrementAccessCount();
+ }
+ return tableBlock;
+ }
+
+ /**
+ * @param absoluteTableIdentifier
+ * @param blockInfo
+ */
+ private void fillSegmentIdToBlockListMap(AbsoluteTableIdentifier absoluteTableIdentifier,
+ BlockInfo blockInfo) {
+ TableSegmentUniqueIdentifier segmentIdentifier =
+ new TableSegmentUniqueIdentifier(absoluteTableIdentifier,
+ blockInfo.getTableBlockInfo().getSegmentId());
+ String uniqueTableSegmentIdentifier = segmentIdentifier.getUniqueTableSegmentIdentifier();
+ List<BlockInfo> blockInfos =
+ segmentIdToBlockListMap.get(uniqueTableSegmentIdentifier);
+ if (null == blockInfos) {
+ Object segmentLockObject = segmentIDLock.get(uniqueTableSegmentIdentifier);
+ if (null == segmentLockObject) {
+ synchronized (segmentIDLock) {
+ segmentLockObject = segmentIDLock.get(uniqueTableSegmentIdentifier);
+ if (null == segmentLockObject) {
+ segmentLockObject = new Object();
+ segmentIDLock.put(uniqueTableSegmentIdentifier, segmentLockObject);
+ }
+ }
+ }
+ synchronized (segmentLockObject) {
+ blockInfos =
+ segmentIdToBlockListMap.get(segmentIdentifier.getUniqueTableSegmentIdentifier());
+ if (null == blockInfos) {
+ blockInfos = new CopyOnWriteArrayList<>();
+ segmentIdToBlockListMap.put(uniqueTableSegmentIdentifier, blockInfos);
+ }
+ blockInfos.add(blockInfo);
+ }
+ } else {
+ blockInfos.add(blockInfo);
+ }
+ }
+
+ /**
+ * The method takes list of tableblocks as input and load them in btree lru cache
+ * and returns the list of data blocks meta
+ *
+ * @param tableBlocksInfos List of unique table blocks
+ * @return List<AbstractIndex>
+ * @throws IndexBuilderException
+ */
+ @Override public List<AbstractIndex> getAll(List<TableBlockUniqueIdentifier> tableBlocksInfos)
+ throws IndexBuilderException {
+ AbstractIndex[] loadedBlock = new AbstractIndex[tableBlocksInfos.size()];
+ int numberOfCores = 1;
+ try {
+ numberOfCores = Integer.parseInt(CarbonProperties.getInstance()
+ .getProperty(CarbonCommonConstants.NUM_CORES,
+ CarbonCommonConstants.NUM_CORES_DEFAULT_VAL));
+ } catch (NumberFormatException e) {
+ numberOfCores = Integer.parseInt(CarbonCommonConstants.NUM_CORES_DEFAULT_VAL);
+ }
+ ExecutorService executor = Executors.newFixedThreadPool(numberOfCores);
+ List<Future<AbstractIndex>> blocksList = new ArrayList<Future<AbstractIndex>>();
+ for (TableBlockUniqueIdentifier tableBlockUniqueIdentifier : tableBlocksInfos) {
+ blocksList.add(executor.submit(new BlockLoaderThread(tableBlockUniqueIdentifier)));
+ }
+ // shutdown the executor gracefully and wait until all the task is finished
+ executor.shutdown();
+ try {
+ executor.awaitTermination(1, TimeUnit.HOURS);
+ } catch (InterruptedException e) {
+ throw new IndexBuilderException(e);
+ }
+ // fill the block which were not loaded before to loaded blocks array
+ fillLoadedBlocks(loadedBlock, blocksList);
+ return Arrays.asList(loadedBlock);
+ }
+
+ private String getLruCacheKey(AbsoluteTableIdentifier absoluteTableIdentifier,
+ BlockInfo blockInfo) {
+ CarbonTableIdentifier carbonTableIdentifier =
+ absoluteTableIdentifier.getCarbonTableIdentifier();
+ return carbonTableIdentifier.getDatabaseName() + CarbonCommonConstants.FILE_SEPARATOR
+ + carbonTableIdentifier.getTableName() + CarbonCommonConstants.UNDERSCORE
+ + carbonTableIdentifier.getTableId() + CarbonCommonConstants.FILE_SEPARATOR + blockInfo
+ .getBlockUniqueName();
+ }
+
+ /**
+ * method returns the B-Tree meta
+ *
+ * @param tableBlockUniqueIdentifier Unique table block info
+ * @return
+ */
+ @Override public AbstractIndex getIfPresent(
+ TableBlockUniqueIdentifier tableBlockUniqueIdentifier) {
+ BlockInfo blockInfo = new BlockInfo(tableBlockUniqueIdentifier.getTableBlockInfo());
+ BlockIndex cacheable = (BlockIndex) lruCache
+ .get(getLruCacheKey(tableBlockUniqueIdentifier.getAbsoluteTableIdentifier(), blockInfo));
+ if (null != cacheable) {
+ cacheable.incrementAccessCount();
+ }
+ return cacheable;
+ }
+
+ /**
+ * the method removes the entry from cache.
+ *
+ * @param tableBlockUniqueIdentifier
+ */
+ @Override public void invalidate(TableBlockUniqueIdentifier tableBlockUniqueIdentifier) {
+ BlockInfo blockInfo = new BlockInfo(tableBlockUniqueIdentifier.getTableBlockInfo());
+ lruCache
+ .remove(getLruCacheKey(tableBlockUniqueIdentifier.getAbsoluteTableIdentifier(), blockInfo));
+ }
+
+ @Override public void clearAccessCount(List<TableBlockUniqueIdentifier> keys) {
+ for (TableBlockUniqueIdentifier tableBlockUniqueIdentifier : keys) {
+ SegmentTaskIndexWrapper cacheable = (SegmentTaskIndexWrapper) lruCache
+ .get(tableBlockUniqueIdentifier.getUniqueTableBlockName());
+ cacheable.clear();
+ }
+ }
+
+ /**
+ * Below method will be used to fill the loaded blocks to the array
+ * which will be used for query execution
+ *
+ * @param loadedBlockArray array of blocks which will be filled
+ * @param blocksList blocks loaded in thread
+ * @throws IndexBuilderException in case of any failure
+ */
+ private void fillLoadedBlocks(AbstractIndex[] loadedBlockArray,
+ List<Future<AbstractIndex>> blocksList) throws IndexBuilderException {
+ int blockCounter = 0;
+ boolean exceptionOccurred = false;
+ Throwable exceptionRef = null;
+ for (int i = 0; i < loadedBlockArray.length; i++) {
+ try {
+ loadedBlockArray[i] = blocksList.get(blockCounter++).get();
+ } catch (Throwable e) {
+ exceptionOccurred = true;
+ exceptionRef = e;
+ }
+ }
+ if (exceptionOccurred) {
+ LOGGER.error("Block B-tree loading failed. Clearing the access count of the loaded blocks.");
+ // in case of any failure clear the access count for the valid loaded blocks
+ clearAccessCountForLoadedBlocks(loadedBlockArray);
+ throw new IndexBuilderException("Block B-tree loading failed", exceptionRef);
+ }
+ }
+
+ /**
+ * This method will clear the access count for the loaded blocks
+ *
+ * @param loadedBlockArray
+ */
+ private void clearAccessCountForLoadedBlocks(AbstractIndex[] loadedBlockArray) {
+ for (int i = 0; i < loadedBlockArray.length; i++) {
+ if (null != loadedBlockArray[i]) {
+ loadedBlockArray[i].clear();
+ }
+ }
+ }
+
+ /**
+ * Thread class which will be used to load the blocks
+ */
+ private class BlockLoaderThread implements Callable<AbstractIndex> {
+ // table block unique identifier
+ private TableBlockUniqueIdentifier tableBlockUniqueIdentifier;
+
+ private BlockLoaderThread(TableBlockUniqueIdentifier tableBlockUniqueIdentifier) {
+ this.tableBlockUniqueIdentifier = tableBlockUniqueIdentifier;
+ }
+
+ @Override public AbstractIndex call() throws Exception {
+ // load and return the loaded blocks
+ return get(tableBlockUniqueIdentifier);
+ }
+ }
+
+ private AbstractIndex loadBlock(TableBlockUniqueIdentifier tableBlockUniqueIdentifier)
+ throws IOException {
+ AbstractIndex tableBlock = new BlockIndex();
+ BlockInfo blockInfo = new BlockInfo(tableBlockUniqueIdentifier.getTableBlockInfo());
+ String lruCacheKey =
+ getLruCacheKey(tableBlockUniqueIdentifier.getAbsoluteTableIdentifier(), blockInfo);
+ checkAndLoadTableBlocks(tableBlock, tableBlockUniqueIdentifier, lruCacheKey);
+ // finally remove the lock object from block info lock as once block is loaded
+ // it will not come inside this if condition
+ blockInfoLock.remove(blockInfo);
+ return tableBlock;
+ }
+
+ /**
+ * This will be used to remove a particular blocks useful in case of
+ * deletion of some of the blocks in case of retention or may be some other
+ * scenario
+ *
+ * @param segmentIds list of table blocks to be removed
+ * @param absoluteTableIdentifier absolute table identifier
+ */
+ public void removeTableBlocks(List<String> segmentIds,
+ AbsoluteTableIdentifier absoluteTableIdentifier) {
+ if (null == segmentIds) {
+ return;
+ }
+ for (String segmentId : segmentIds) {
+ TableSegmentUniqueIdentifier tableSegmentUniqueIdentifier =
+ new TableSegmentUniqueIdentifier(absoluteTableIdentifier, segmentId);
+ List<BlockInfo> blockInfos = segmentIdToBlockListMap
+ .remove(tableSegmentUniqueIdentifier.getUniqueTableSegmentIdentifier());
+ if (null != blockInfos) {
+ for (BlockInfo blockInfo : blockInfos) {
+ String lruCacheKey = getLruCacheKey(absoluteTableIdentifier, blockInfo);
+ lruCache.remove(lruCacheKey);
+ }
+ }
+ }
+ }
+
+ /**
+ * remove TableBlocks executer level If Horizontal Compaction Done
+ * @param queryModel
+ */
+ public void removeTableBlocksIfHorizontalCompactionDone(QueryModel queryModel) {
+ // get the invalid segments blocks details
+ Map<String, UpdateVO> invalidBlocksVO = queryModel.getInvalidBlockVOForSegmentId();
+ if (!invalidBlocksVO.isEmpty()) {
+ UpdateVO updateMetadata;
+ Iterator<Map.Entry<String, UpdateVO>> itr = invalidBlocksVO.entrySet().iterator();
+ String blockTimestamp = null;
+ while (itr.hasNext()) {
+ Map.Entry<String, UpdateVO> entry = itr.next();
+ TableSegmentUniqueIdentifier tableSegmentUniqueIdentifier =
+ new TableSegmentUniqueIdentifier(queryModel.getAbsoluteTableIdentifier(),
+ entry.getKey());
+ List<BlockInfo> blockInfos = segmentIdToBlockListMap
+ .get(tableSegmentUniqueIdentifier.getUniqueTableSegmentIdentifier());
+ if (null != blockInfos) {
+ for (BlockInfo blockInfo : blockInfos) {
+ // reading the updated block names from status manager instance
+ blockTimestamp = blockInfo.getBlockUniqueName()
+ .substring(blockInfo.getBlockUniqueName().lastIndexOf('-') + 1,
+ blockInfo.getBlockUniqueName().length());
+ updateMetadata = entry.getValue();
+ if (CarbonUpdateUtil.isMaxQueryTimeoutExceeded(Long.parseLong(blockTimestamp))) {
+ Long blockTimeStamp = Long.parseLong(blockTimestamp);
+ if (blockTimeStamp > updateMetadata.getFactTimestamp() && (
+ updateMetadata.getUpdateDeltaStartTimestamp() != null
+ && blockTimeStamp < updateMetadata.getUpdateDeltaStartTimestamp())) {
+ String lruCacheKey =
+ getLruCacheKey(queryModel.getAbsoluteTableIdentifier(), blockInfo);
+ lruCache.remove(lruCacheKey);
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+}