1112

Build # 1112

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/7041b2007eb5c1006444e41d675ec0a256df36ac">7041b2007<a href="https://github.com/apache/carbondata/commit/7041b2007eb5c1006444e41d675ec0a256df36ac">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/9f752b74081806d5d4f5aece1558b968006b8c7c">9f752b740</a>">9f752b740</a><a href="https://github.com/apache/carbondata/commit/7041b2007eb5c1006444e41d675ec0a256df36ac"> into 8c49e5b42">8c49e5b42</a>

Run Details

4 of 4 new or added lines in 1 file covered. (100.0%)

60552 of 75153 relevant lines covered (80.57%)

1.06 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.71

/core/src/main/java/org/apache/carbondata/core/datastore/block/SegmentPropertiesAndSchemaHolder.java

/*
 * 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.block;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.indexstore.schema.CarbonRowSchema;
import org.apache.carbondata.core.indexstore.schema.SchemaGenerator;
import org.apache.carbondata.core.memory.MemoryException;
import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;

import org.apache.log4j.Logger;

/**
 * Singleton class which will help in creating the segment properties
 */
public class SegmentPropertiesAndSchemaHolder {

  /**
   * Logger
   */
  private static final Logger LOGGER =
      LogServiceFactory.getLogService(SegmentPropertiesAndSchemaHolder.class.getName());
  /**
   * SegmentPropertiesAndSchemaHolder instance
   */
  private static final SegmentPropertiesAndSchemaHolder INSTANCE =
      new SegmentPropertiesAndSchemaHolder();
  /**
   * object level lock
   */
  private static final Object lock = new Object();
  /**
   * counter for maintaining the index of segmentProperties
   */
  private static final AtomicInteger segmentPropertiesIndexCounter = new AtomicInteger(0);
  /**
   * holds segmentPropertiesWrapper to segment ID and segmentProperties Index wrapper mapping.
   * Will be used while invaliding a segment and drop table
   */
  private Map<SegmentPropertiesWrapper, SegmentIdAndSegmentPropertiesIndexWrapper>
      segmentPropWrapperToSegmentSetMap = new ConcurrentHashMap<>();
  /**
   * reverse mapping for segmentProperties index to segmentPropertiesWrapper
   */
  private static Map<Integer, SegmentPropertiesWrapper> indexToSegmentPropertiesWrapperMapping =
      new ConcurrentHashMap<>();
  /**
   * Map to be used for table level locking while populating segmentProperties
   */
  private Map<String, Object> absoluteTableIdentifierByteMap = new ConcurrentHashMap<>();

  /**
   * private constructor for singleton instance
   */
  private SegmentPropertiesAndSchemaHolder() {

  }

  public static SegmentPropertiesAndSchemaHolder getInstance() {
    return INSTANCE;
  }

  /**
   * Method to add the segment properties and avoid construction of new segment properties until
   * the schema is not modified
   *
   * @param carbonTable
   * @param columnsInTable
   * @param columnCardinality
   * @param segmentId
   */
  public int addSegmentProperties(CarbonTable carbonTable,
      List<ColumnSchema> columnsInTable, int[] columnCardinality, String segmentId) {
    SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper segmentPropertiesWrapper =
        new SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper(carbonTable,
            columnsInTable, columnCardinality);
    SegmentIdAndSegmentPropertiesIndexWrapper segmentIdSetAndIndexWrapper =
        this.segmentPropWrapperToSegmentSetMap.get(segmentPropertiesWrapper);
    if (null == segmentIdSetAndIndexWrapper) {
      synchronized (getOrCreateTableLock(carbonTable.getAbsoluteTableIdentifier())) {
        segmentIdSetAndIndexWrapper =
            this.segmentPropWrapperToSegmentSetMap.get(segmentPropertiesWrapper);
        if (null == segmentIdSetAndIndexWrapper) {
          // create new segmentProperties
          segmentPropertiesWrapper.initSegmentProperties();
          segmentPropertiesWrapper.addMinMaxColumns(carbonTable);
          int segmentPropertiesIndex = segmentPropertiesIndexCounter.incrementAndGet();
          indexToSegmentPropertiesWrapperMapping
              .put(segmentPropertiesIndex, segmentPropertiesWrapper);
          LOGGER.info("Constructing new SegmentProperties for table: " + carbonTable
              .getCarbonTableIdentifier().getTableUniqueName()
              + ". Current size of segment properties" + " holder list is: "
              + indexToSegmentPropertiesWrapperMapping.size());
          // populate the SegmentIdAndSegmentPropertiesIndexWrapper to maintain the set of segments
          // having same SegmentPropertiesWrapper instance this will used to decide during the
          // tblSegmentsProperties map clean-up for the invalid segments
          segmentIdSetAndIndexWrapper =
              new SegmentIdAndSegmentPropertiesIndexWrapper(segmentId, segmentPropertiesIndex);
          segmentPropWrapperToSegmentSetMap
              .put(segmentPropertiesWrapper, segmentIdSetAndIndexWrapper);
        }
      }
    } else {
      synchronized (getOrCreateTableLock(carbonTable.getAbsoluteTableIdentifier())) {
        segmentIdSetAndIndexWrapper.addSegmentId(segmentId);
        indexToSegmentPropertiesWrapperMapping
            .get(segmentIdSetAndIndexWrapper.getSegmentPropertiesIndex())
            .addMinMaxColumns(carbonTable);
      }
    }
    return segmentIdSetAndIndexWrapper.getSegmentPropertiesIndex();
  }

  /**
   * Method to create table Level lock
   *
   * @param absoluteTableIdentifier
   * @return
   */
  private Object getOrCreateTableLock(AbsoluteTableIdentifier absoluteTableIdentifier) {
    Object tableLock = absoluteTableIdentifierByteMap
        .get(absoluteTableIdentifier.getCarbonTableIdentifier().getTableUniqueName());
    if (null == tableLock) {
      synchronized (lock) {
        tableLock = absoluteTableIdentifierByteMap
            .get(absoluteTableIdentifier.getCarbonTableIdentifier().getTableUniqueName());
        if (null == tableLock) {
          tableLock = new Object();
          absoluteTableIdentifierByteMap
              .put(absoluteTableIdentifier.getCarbonTableIdentifier().getTableUniqueName(),
                  tableLock);
        }
      }
    }
    return tableLock;
  }

  /**
   * Method to get the segment properties from given index
   *
   * @param segmentPropertiesIndex
   * @return
   */
  public SegmentProperties getSegmentProperties(int segmentPropertiesIndex) {
    SegmentPropertiesWrapper segmentPropertiesWrapper =
        getSegmentPropertiesWrapper(segmentPropertiesIndex);
    if (null != segmentPropertiesWrapper) {
      return segmentPropertiesWrapper.getSegmentProperties();
    }
    return null;
  }

  /**
   * Method to get the segment properties from given index
   *
   * @param segmentPropertiesWrapperIndex
   * @return
   */
  public SegmentPropertiesWrapper getSegmentPropertiesWrapper(int segmentPropertiesWrapperIndex) {
    return indexToSegmentPropertiesWrapperMapping.get(segmentPropertiesWrapperIndex);
  }

  /**
   * This method will remove the segment properties from the map on drop table
   *
   * @param absoluteTableIdentifier
   */
  public void invalidate(AbsoluteTableIdentifier absoluteTableIdentifier) {
    List<SegmentPropertiesWrapper> segmentPropertiesWrappersToBeRemoved = new ArrayList<>();
    // remove segmentProperties wrapper entries from the copyOnWriteArrayList
    for (Map.Entry<SegmentPropertiesWrapper, SegmentIdAndSegmentPropertiesIndexWrapper> entry :
          segmentPropWrapperToSegmentSetMap.entrySet()) {
      SegmentPropertiesWrapper segmentPropertiesWrapper = entry.getKey();
      if (segmentPropertiesWrapper.getTableIdentifier().getCarbonTableIdentifier()
          .getTableUniqueName()
          .equals(absoluteTableIdentifier.getCarbonTableIdentifier().getTableUniqueName())) {
        SegmentIdAndSegmentPropertiesIndexWrapper value = entry.getValue();
        // remove from the reverse mapping map
        indexToSegmentPropertiesWrapperMapping.remove(value.getSegmentPropertiesIndex());
        segmentPropertiesWrappersToBeRemoved.add(segmentPropertiesWrapper);
      }
    }
    // remove all the segmentPropertiesWrapper entries from map
    for (SegmentPropertiesWrapper segmentPropertiesWrapper : segmentPropertiesWrappersToBeRemoved) {
      segmentPropWrapperToSegmentSetMap.remove(segmentPropertiesWrapper);
    }
    // remove the table lock
    absoluteTableIdentifierByteMap
        .remove(absoluteTableIdentifier.getCarbonTableIdentifier().getTableUniqueName());
  }

  /**
   * Method to remove the given segment ID
   *
   * @param segmentId
   * @param segmentPropertiesIndex
   * @param clearSegmentWrapperFromMap flag to specify whether to clear segmentPropertiesWrapper
   *                                   from Map if all the segment's using it have become stale
   */
  public void invalidate(String segmentId, int segmentPropertiesIndex,
      boolean clearSegmentWrapperFromMap) {
    SegmentPropertiesWrapper segmentPropertiesWrapper =
        indexToSegmentPropertiesWrapperMapping.get(segmentPropertiesIndex);
    if (null != segmentPropertiesWrapper) {
      SegmentIdAndSegmentPropertiesIndexWrapper segmentIdAndSegmentPropertiesIndexWrapper =
          segmentPropWrapperToSegmentSetMap.get(segmentPropertiesWrapper);
      synchronized (getOrCreateTableLock(segmentPropertiesWrapper.getTableIdentifier())) {
        segmentIdAndSegmentPropertiesIndexWrapper.removeSegmentId(segmentId);
      }
      // if after removal of given SegmentId, the segmentIdSet becomes empty that means this
      // segmentPropertiesWrapper is not getting used at all. In that case this object can be
      // removed from all the holders
      if (clearSegmentWrapperFromMap && segmentIdAndSegmentPropertiesIndexWrapper.segmentIdSet
          .isEmpty()) {
        indexToSegmentPropertiesWrapperMapping.remove(segmentPropertiesIndex);
        segmentPropWrapperToSegmentSetMap.remove(segmentPropertiesWrapper);
      } else if (!clearSegmentWrapperFromMap
          && segmentIdAndSegmentPropertiesIndexWrapper.segmentIdSet.isEmpty()) {
        // min max columns can very when cache is modified. So even though entry is not required
        // to be deleted from map clear the column cache so that it can filled again
        segmentPropertiesWrapper.clear();
        LOGGER.info("cleared min max for segmentProperties at index: " + segmentPropertiesIndex);
      }
    }
  }

  /**
   * add segmentId at given segmentPropertyIndex
   * Note: This method is getting used in extension with other features. Please do not remove
   *
   * @param segmentPropertiesIndex
   * @param segmentId
   */
  public void addSegmentId(int segmentPropertiesIndex, String segmentId) {
    SegmentPropertiesWrapper segmentPropertiesWrapper =
        indexToSegmentPropertiesWrapperMapping.get(segmentPropertiesIndex);
    if (null != segmentPropertiesWrapper) {
      SegmentIdAndSegmentPropertiesIndexWrapper segmentIdAndSegmentPropertiesIndexWrapper =
          segmentPropWrapperToSegmentSetMap.get(segmentPropertiesWrapper);
      synchronized (getOrCreateTableLock(segmentPropertiesWrapper.getTableIdentifier())) {
        segmentIdAndSegmentPropertiesIndexWrapper.addSegmentId(segmentId);
      }
    }
  }

  /**
   * This class wraps tableIdentifier, columnsInTable and columnCardinality as a key to determine
   * whether the SegmentProperties object can be reused.
   */
  public static class SegmentPropertiesWrapper {

    private static final Object taskSchemaLock = new Object();
    private static final Object fileFooterSchemaLock = new Object();

    private AbsoluteTableIdentifier tableIdentifier;
    private List<ColumnSchema> columnsInTable;
    private int[] columnCardinality;
    private SegmentProperties segmentProperties;
    private List<CarbonColumn> minMaxCacheColumns;
    private CarbonRowSchema[] taskSummarySchema;
    // same variable can be used for block and blocklet schema because at any given cache_level
    // with either block or blocklet and whenever cache_level is changed the cache and its
    // corresponding segmentProperties is flushed
    private CarbonRowSchema[] fileFooterEntrySchema;

    public SegmentPropertiesWrapper(CarbonTable carbonTable,
        List<ColumnSchema> columnsInTable, int[] columnCardinality) {
      this.tableIdentifier = carbonTable.getAbsoluteTableIdentifier();
      this.columnsInTable = columnsInTable;
      this.columnCardinality = columnCardinality;
    }

    public void initSegmentProperties() {
      segmentProperties = new SegmentProperties(columnsInTable, columnCardinality);
    }

    public void addMinMaxColumns(CarbonTable carbonTable) {
      if (null == minMaxCacheColumns || minMaxCacheColumns.isEmpty()) {
        minMaxCacheColumns = carbonTable.getMinMaxCacheColumns(segmentProperties);
      }
    }

    /**
     * clear required fields
     */
    public void clear() {
      if (null != minMaxCacheColumns) {
        minMaxCacheColumns.clear();
      }
      taskSummarySchema = null;
      fileFooterEntrySchema = null;
    }

    @Override public boolean equals(Object obj) {
      if (!(obj instanceof SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper)) {
        return false;
      }
      SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper other =
          (SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper) obj;
      return tableIdentifier.equals(other.tableIdentifier) && columnsInTable
          .equals(other.columnsInTable) && Arrays
          .equals(columnCardinality, other.columnCardinality);
    }

    @Override public int hashCode() {
      return tableIdentifier.hashCode() + columnsInTable.hashCode() + Arrays
          .hashCode(columnCardinality);
    }

    public AbsoluteTableIdentifier getTableIdentifier() {
      return tableIdentifier;
    }

    public SegmentProperties getSegmentProperties() {
      return segmentProperties;
    }

    public List<ColumnSchema> getColumnsInTable() {
      return columnsInTable;
    }

    public int[] getColumnCardinality() {
      return columnCardinality;
    }

    public CarbonRowSchema[] getTaskSummarySchema(boolean storeBlockletCount,
        boolean filePathToBeStored) throws MemoryException {
      if (null == taskSummarySchema) {
        synchronized (taskSchemaLock) {
          if (null == taskSummarySchema) {
            taskSummarySchema = SchemaGenerator
                .createTaskSummarySchema(segmentProperties, minMaxCacheColumns, storeBlockletCount,
                    filePathToBeStored);
          }
        }
      }
      return taskSummarySchema;
    }

    public CarbonRowSchema[] getBlockFileFooterEntrySchema() {
      return getOrCreateFileFooterEntrySchema(true);
    }

    public CarbonRowSchema[] getBlockletFileFooterEntrySchema() {
      return getOrCreateFileFooterEntrySchema(false);
    }

    public List<CarbonColumn> getMinMaxCacheColumns() {
      return minMaxCacheColumns;
    }

    private CarbonRowSchema[] getOrCreateFileFooterEntrySchema(boolean isCacheLevelBlock) {
      if (null == fileFooterEntrySchema) {
        synchronized (fileFooterSchemaLock) {
          if (null == fileFooterEntrySchema) {
            if (isCacheLevelBlock) {
              fileFooterEntrySchema =
                  SchemaGenerator.createBlockSchema(segmentProperties, minMaxCacheColumns);
            } else {
              fileFooterEntrySchema =
                  SchemaGenerator.createBlockletSchema(segmentProperties, minMaxCacheColumns);
            }
          }
        }
      }
      return fileFooterEntrySchema;
    }
  }

  /**
   * holder for segmentId and segmentPropertiesIndex
   */
  public static class SegmentIdAndSegmentPropertiesIndexWrapper {

    /**
     * set holding all unique segment Id's using the same segmentProperties
     */
    private Set<String> segmentIdSet;
    /**
     * index which maps to segmentPropertiesWrpper Index from where segmentProperties
     * can be retrieved
     */
    private int segmentPropertiesIndex;

    public SegmentIdAndSegmentPropertiesIndexWrapper(String segmentId, int segmentPropertiesIndex) {
      segmentIdSet = new HashSet<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);
      addSegmentId(segmentId);
      this.segmentPropertiesIndex = segmentPropertiesIndex;
    }

    public void addSegmentId(String segmentId) {
      segmentIdSet.add(segmentId);
    }

    public void removeSegmentId(String segmentId) {
      segmentIdSet.remove(segmentId);
    }

    public int getSegmentPropertiesIndex() {
      return segmentPropertiesIndex;
    }
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17	package org.apache.carbondata.core.datastore.block;
18
19	import java.util.ArrayList;
20	import java.util.Arrays;
21	import java.util.HashSet;
22	import java.util.List;
23	import java.util.Map;
24	import java.util.Set;
25	import java.util.concurrent.ConcurrentHashMap;
26	import java.util.concurrent.atomic.AtomicInteger;
27
28	import org.apache.carbondata.common.logging.LogServiceFactory;
29	import org.apache.carbondata.core.constants.CarbonCommonConstants;
30	import org.apache.carbondata.core.indexstore.schema.CarbonRowSchema;
31	import org.apache.carbondata.core.indexstore.schema.SchemaGenerator;
32	import org.apache.carbondata.core.memory.MemoryException;
33	import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
34	import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
35	import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn;
36	import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema;
37
38	import org.apache.log4j.Logger;
39
40	/**
41	* Singleton class which will help in creating the segment properties
42	*/
43	public class SegmentPropertiesAndSchemaHolder {
44
45	/**
46	* Logger
47	*/
48	private static final Logger LOGGER =	1✔
49	LogServiceFactory.getLogService(SegmentPropertiesAndSchemaHolder.class.getName());	1✔
50	/**
51	* SegmentPropertiesAndSchemaHolder instance
52	*/
53	private static final SegmentPropertiesAndSchemaHolder INSTANCE =	1✔
54	new SegmentPropertiesAndSchemaHolder();
55	/**
56	* object level lock
57	*/
58	private static final Object lock = new Object();	1✔
59	/**
60	* counter for maintaining the index of segmentProperties
61	*/
62	private static final AtomicInteger segmentPropertiesIndexCounter = new AtomicInteger(0);	1✔
63	/**
64	* holds segmentPropertiesWrapper to segment ID and segmentProperties Index wrapper mapping.
65	* Will be used while invaliding a segment and drop table
66	*/
67	private Map<SegmentPropertiesWrapper, SegmentIdAndSegmentPropertiesIndexWrapper>	1✔
68	segmentPropWrapperToSegmentSetMap = new ConcurrentHashMap<>();
69	/**
70	* reverse mapping for segmentProperties index to segmentPropertiesWrapper
71	*/
72	private static Map<Integer, SegmentPropertiesWrapper> indexToSegmentPropertiesWrapperMapping =	1✔
73	new ConcurrentHashMap<>();
74	/**
75	* Map to be used for table level locking while populating segmentProperties
76	*/
77	private Map<String, Object> absoluteTableIdentifierByteMap = new ConcurrentHashMap<>();	1✔
78
79	/**
80	* private constructor for singleton instance
81	*/
82	private SegmentPropertiesAndSchemaHolder() {	1✔
83
84	}	1✔
85
86	public static SegmentPropertiesAndSchemaHolder getInstance() {
87	return INSTANCE;	1✔
88	}
89
90	/**
91	* Method to add the segment properties and avoid construction of new segment properties until
92	* the schema is not modified
93	*
94	* @param carbonTable
95	* @param columnsInTable
96	* @param columnCardinality
97	* @param segmentId
98	*/
99	public int addSegmentProperties(CarbonTable carbonTable,
100	List<ColumnSchema> columnsInTable, int[] columnCardinality, String segmentId) {
101	SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper segmentPropertiesWrapper =	1✔
102	new SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper(carbonTable,
103	columnsInTable, columnCardinality);
104	SegmentIdAndSegmentPropertiesIndexWrapper segmentIdSetAndIndexWrapper =	1✔
105	this.segmentPropWrapperToSegmentSetMap.get(segmentPropertiesWrapper);	1✔
106	if (null == segmentIdSetAndIndexWrapper) {	1✔
107	synchronized (getOrCreateTableLock(carbonTable.getAbsoluteTableIdentifier())) {	1✔
108	segmentIdSetAndIndexWrapper =	1✔
109	this.segmentPropWrapperToSegmentSetMap.get(segmentPropertiesWrapper);	1✔
110	if (null == segmentIdSetAndIndexWrapper) {	1✔
111	// create new segmentProperties
112	segmentPropertiesWrapper.initSegmentProperties();	1✔
113	segmentPropertiesWrapper.addMinMaxColumns(carbonTable);	1✔
114	int segmentPropertiesIndex = segmentPropertiesIndexCounter.incrementAndGet();	1✔
115	indexToSegmentPropertiesWrapperMapping	1✔
116	.put(segmentPropertiesIndex, segmentPropertiesWrapper);	1✔
117	LOGGER.info("Constructing new SegmentProperties for table: " + carbonTable	1✔
118	.getCarbonTableIdentifier().getTableUniqueName()	1✔
119	+ ". Current size of segment properties" + " holder list is: "
120	+ indexToSegmentPropertiesWrapperMapping.size());	1✔
121	// populate the SegmentIdAndSegmentPropertiesIndexWrapper to maintain the set of segments
122	// having same SegmentPropertiesWrapper instance this will used to decide during the
123	// tblSegmentsProperties map clean-up for the invalid segments
124	segmentIdSetAndIndexWrapper =	1✔
125	new SegmentIdAndSegmentPropertiesIndexWrapper(segmentId, segmentPropertiesIndex);
126	segmentPropWrapperToSegmentSetMap	1✔
127	.put(segmentPropertiesWrapper, segmentIdSetAndIndexWrapper);	1✔
128	}
129	}	1✔
130	} else {
131	synchronized (getOrCreateTableLock(carbonTable.getAbsoluteTableIdentifier())) {	1✔
132	segmentIdSetAndIndexWrapper.addSegmentId(segmentId);	1✔
133	indexToSegmentPropertiesWrapperMapping	1✔
134	.get(segmentIdSetAndIndexWrapper.getSegmentPropertiesIndex())	1✔
135	.addMinMaxColumns(carbonTable);	1✔
136	}	1✔
137	}
138	return segmentIdSetAndIndexWrapper.getSegmentPropertiesIndex();	1✔
139	}
140
141	/**
142	* Method to create table Level lock
143	*
144	* @param absoluteTableIdentifier
145	* @return
146	*/
147	private Object getOrCreateTableLock(AbsoluteTableIdentifier absoluteTableIdentifier) {
148	Object tableLock = absoluteTableIdentifierByteMap	1✔
149	.get(absoluteTableIdentifier.getCarbonTableIdentifier().getTableUniqueName());	1✔
150	if (null == tableLock) {	1✔
151	synchronized (lock) {	1✔
152	tableLock = absoluteTableIdentifierByteMap	1✔
153	.get(absoluteTableIdentifier.getCarbonTableIdentifier().getTableUniqueName());	1✔
154	if (null == tableLock) {	1✔
155	tableLock = new Object();	1✔
156	absoluteTableIdentifierByteMap	1✔
157	.put(absoluteTableIdentifier.getCarbonTableIdentifier().getTableUniqueName(),	1✔
158	tableLock);
159	}
160	}	1✔
161	}
162	return tableLock;	1✔
163	}
164
165	/**
166	* Method to get the segment properties from given index
167	*
168	* @param segmentPropertiesIndex
169	* @return
170	*/
171	public SegmentProperties getSegmentProperties(int segmentPropertiesIndex) {
172	SegmentPropertiesWrapper segmentPropertiesWrapper =	1✔
173	getSegmentPropertiesWrapper(segmentPropertiesIndex);	1✔
174	if (null != segmentPropertiesWrapper) {	1✔
175	return segmentPropertiesWrapper.getSegmentProperties();	1✔
176	}
177	return null;	×
178	}
179
180	/**
181	* Method to get the segment properties from given index
182	*
183	* @param segmentPropertiesWrapperIndex
184	* @return
185	*/
186	public SegmentPropertiesWrapper getSegmentPropertiesWrapper(int segmentPropertiesWrapperIndex) {
187	return indexToSegmentPropertiesWrapperMapping.get(segmentPropertiesWrapperIndex);	1✔
188	}
189
190	/**
191	* This method will remove the segment properties from the map on drop table
192	*
193	* @param absoluteTableIdentifier
194	*/
195	public void invalidate(AbsoluteTableIdentifier absoluteTableIdentifier) {
196	List<SegmentPropertiesWrapper> segmentPropertiesWrappersToBeRemoved = new ArrayList<>();	1✔
197	// remove segmentProperties wrapper entries from the copyOnWriteArrayList
198	for (Map.Entry<SegmentPropertiesWrapper, SegmentIdAndSegmentPropertiesIndexWrapper> entry :
199	segmentPropWrapperToSegmentSetMap.entrySet()) {	1✔
200	SegmentPropertiesWrapper segmentPropertiesWrapper = entry.getKey();	1✔
201	if (segmentPropertiesWrapper.getTableIdentifier().getCarbonTableIdentifier()	1✔
202	.getTableUniqueName()	1✔
203	.equals(absoluteTableIdentifier.getCarbonTableIdentifier().getTableUniqueName())) {	1✔
204	SegmentIdAndSegmentPropertiesIndexWrapper value = entry.getValue();	×
205	// remove from the reverse mapping map
206	indexToSegmentPropertiesWrapperMapping.remove(value.getSegmentPropertiesIndex());	×
207	segmentPropertiesWrappersToBeRemoved.add(segmentPropertiesWrapper);	×
208	}
209	}	1✔
210	// remove all the segmentPropertiesWrapper entries from map
211	for (SegmentPropertiesWrapper segmentPropertiesWrapper : segmentPropertiesWrappersToBeRemoved) {	1✔
212	segmentPropWrapperToSegmentSetMap.remove(segmentPropertiesWrapper);	×
213	}	×
214	// remove the table lock
215	absoluteTableIdentifierByteMap	1✔
216	.remove(absoluteTableIdentifier.getCarbonTableIdentifier().getTableUniqueName());	1✔
217	}	1✔
218
219	/**
220	* Method to remove the given segment ID
221	*
222	* @param segmentId
223	* @param segmentPropertiesIndex
224	* @param clearSegmentWrapperFromMap flag to specify whether to clear segmentPropertiesWrapper
225	* from Map if all the segment's using it have become stale
226	*/
227	public void invalidate(String segmentId, int segmentPropertiesIndex,
228	boolean clearSegmentWrapperFromMap) {
229	SegmentPropertiesWrapper segmentPropertiesWrapper =	1✔
230	indexToSegmentPropertiesWrapperMapping.get(segmentPropertiesIndex);	1✔
231	if (null != segmentPropertiesWrapper) {	1✔
232	SegmentIdAndSegmentPropertiesIndexWrapper segmentIdAndSegmentPropertiesIndexWrapper =	1✔
233	segmentPropWrapperToSegmentSetMap.get(segmentPropertiesWrapper);	1✔
234	synchronized (getOrCreateTableLock(segmentPropertiesWrapper.getTableIdentifier())) {	1✔
235	segmentIdAndSegmentPropertiesIndexWrapper.removeSegmentId(segmentId);	1✔
236	}	1✔
237	// if after removal of given SegmentId, the segmentIdSet becomes empty that means this
238	// segmentPropertiesWrapper is not getting used at all. In that case this object can be
239	// removed from all the holders
240	if (clearSegmentWrapperFromMap && segmentIdAndSegmentPropertiesIndexWrapper.segmentIdSet	1✔
241	.isEmpty()) {	1✔
242	indexToSegmentPropertiesWrapperMapping.remove(segmentPropertiesIndex);	1✔
243	segmentPropWrapperToSegmentSetMap.remove(segmentPropertiesWrapper);	1✔
244	} else if (!clearSegmentWrapperFromMap	1✔
245	&& segmentIdAndSegmentPropertiesIndexWrapper.segmentIdSet.isEmpty()) {	×
246	// min max columns can very when cache is modified. So even though entry is not required
247	// to be deleted from map clear the column cache so that it can filled again
248	segmentPropertiesWrapper.clear();	×
249	LOGGER.info("cleared min max for segmentProperties at index: " + segmentPropertiesIndex);	×
250	}
251	}
252	}	1✔
253
254	/**
255	* add segmentId at given segmentPropertyIndex
256	* Note: This method is getting used in extension with other features. Please do not remove
257	*
258	* @param segmentPropertiesIndex
259	* @param segmentId
260	*/
261	public void addSegmentId(int segmentPropertiesIndex, String segmentId) {
262	SegmentPropertiesWrapper segmentPropertiesWrapper =	×
263	indexToSegmentPropertiesWrapperMapping.get(segmentPropertiesIndex);	×
264	if (null != segmentPropertiesWrapper) {	×
265	SegmentIdAndSegmentPropertiesIndexWrapper segmentIdAndSegmentPropertiesIndexWrapper =	×
266	segmentPropWrapperToSegmentSetMap.get(segmentPropertiesWrapper);	×
267	synchronized (getOrCreateTableLock(segmentPropertiesWrapper.getTableIdentifier())) {	×
268	segmentIdAndSegmentPropertiesIndexWrapper.addSegmentId(segmentId);	×
269	}	×
270	}
271	}	×
272
273	/**
274	* This class wraps tableIdentifier, columnsInTable and columnCardinality as a key to determine
275	* whether the SegmentProperties object can be reused.
276	*/
277	public static class SegmentPropertiesWrapper {
278
279	private static final Object taskSchemaLock = new Object();	1✔
280	private static final Object fileFooterSchemaLock = new Object();	1✔
281
282	private AbsoluteTableIdentifier tableIdentifier;
283	private List<ColumnSchema> columnsInTable;
284	private int[] columnCardinality;
285	private SegmentProperties segmentProperties;
286	private List<CarbonColumn> minMaxCacheColumns;
287	private CarbonRowSchema[] taskSummarySchema;
288	// same variable can be used for block and blocklet schema because at any given cache_level
289	// with either block or blocklet and whenever cache_level is changed the cache and its
290	// corresponding segmentProperties is flushed
291	private CarbonRowSchema[] fileFooterEntrySchema;
292
293	public SegmentPropertiesWrapper(CarbonTable carbonTable,
294	List<ColumnSchema> columnsInTable, int[] columnCardinality) {	1✔
295	this.tableIdentifier = carbonTable.getAbsoluteTableIdentifier();	1✔
296	this.columnsInTable = columnsInTable;	1✔
297	this.columnCardinality = columnCardinality;	1✔
298	}	1✔
299
300	public void initSegmentProperties() {
301	segmentProperties = new SegmentProperties(columnsInTable, columnCardinality);	1✔
302	}	1✔
303
304	public void addMinMaxColumns(CarbonTable carbonTable) {
305	if (null == minMaxCacheColumns \|\| minMaxCacheColumns.isEmpty()) {	1✔
306	minMaxCacheColumns = carbonTable.getMinMaxCacheColumns(segmentProperties);	1✔
307	}
308	}	1✔
309
310	/**
311	* clear required fields
312	*/
313	public void clear() {
314	if (null != minMaxCacheColumns) {	×
315	minMaxCacheColumns.clear();	×
316	}
317	taskSummarySchema = null;	×
318	fileFooterEntrySchema = null;	×
319	}	×
320
321	@Override public boolean equals(Object obj) {
322	if (!(obj instanceof SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper)) {	1✔
323	return false;	×
324	}
325	SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper other =	1✔
326	(SegmentPropertiesAndSchemaHolder.SegmentPropertiesWrapper) obj;
327	return tableIdentifier.equals(other.tableIdentifier) && columnsInTable	1✔
328	.equals(other.columnsInTable) && Arrays	1✔
329	.equals(columnCardinality, other.columnCardinality);	1✔
330	}
331
332	@Override public int hashCode() {
333	return tableIdentifier.hashCode() + columnsInTable.hashCode() + Arrays	1✔
334	.hashCode(columnCardinality);	1✔
335	}
336
337	public AbsoluteTableIdentifier getTableIdentifier() {
338	return tableIdentifier;	1✔
339	}
340
341	public SegmentProperties getSegmentProperties() {
342	return segmentProperties;	1✔
343	}
344
345	public List<ColumnSchema> getColumnsInTable() {
346	return columnsInTable;	1✔
347	}
348
349	public int[] getColumnCardinality() {
350	return columnCardinality;	1✔
351	}
352
353	public CarbonRowSchema[] getTaskSummarySchema(boolean storeBlockletCount,
354	boolean filePathToBeStored) throws MemoryException {
355	if (null == taskSummarySchema) {	1✔
356	synchronized (taskSchemaLock) {	1✔
357	if (null == taskSummarySchema) {	1✔
358	taskSummarySchema = SchemaGenerator	1✔
359	.createTaskSummarySchema(segmentProperties, minMaxCacheColumns, storeBlockletCount,	1✔
360	filePathToBeStored);
361	}
362	}	1✔
363	}
364	return taskSummarySchema;	1✔
365	}
366
367	public CarbonRowSchema[] getBlockFileFooterEntrySchema() {
368	return getOrCreateFileFooterEntrySchema(true);	1✔
369	}
370
371	public CarbonRowSchema[] getBlockletFileFooterEntrySchema() {
372	return getOrCreateFileFooterEntrySchema(false);	1✔
373	}
374
375	public List<CarbonColumn> getMinMaxCacheColumns() {
376	return minMaxCacheColumns;	1✔
377	}
378
379	private CarbonRowSchema[] getOrCreateFileFooterEntrySchema(boolean isCacheLevelBlock) {
380	if (null == fileFooterEntrySchema) {	1✔
381	synchronized (fileFooterSchemaLock) {	1✔
382	if (null == fileFooterEntrySchema) {	1✔
383	if (isCacheLevelBlock) {	1✔
384	fileFooterEntrySchema =	1✔
385	SchemaGenerator.createBlockSchema(segmentProperties, minMaxCacheColumns);	1✔
386	} else {
387	fileFooterEntrySchema =	1✔
388	SchemaGenerator.createBlockletSchema(segmentProperties, minMaxCacheColumns);	1✔
389	}
390	}
391	}	1✔
392	}
393	return fileFooterEntrySchema;	1✔
394	}
395	}
396
397	/**
398	* holder for segmentId and segmentPropertiesIndex
399	*/
400	public static class SegmentIdAndSegmentPropertiesIndexWrapper {
401
402	/**
403	* set holding all unique segment Id's using the same segmentProperties
404	*/
405	private Set<String> segmentIdSet;
406	/**
407	* index which maps to segmentPropertiesWrpper Index from where segmentProperties
408	* can be retrieved
409	*/
410	private int segmentPropertiesIndex;
411
412	public SegmentIdAndSegmentPropertiesIndexWrapper(String segmentId, int segmentPropertiesIndex) {	1✔
413	segmentIdSet = new HashSet<>(CarbonCommonConstants.DEFAULT_COLLECTION_SIZE);	1✔
414	addSegmentId(segmentId);	1✔
415	this.segmentPropertiesIndex = segmentPropertiesIndex;	1✔
416	}	1✔
417
418	public void addSegmentId(String segmentId) {
419	segmentIdSet.add(segmentId);	1✔
420	}	1✔
421
422	public void removeSegmentId(String segmentId) {
423	segmentIdSet.remove(segmentId);	1✔
424	}	1✔
425
426	public int getSegmentPropertiesIndex() {
427	return segmentPropertiesIndex;	1✔
428	}
429	}
430	}

apache / carbondata / 1112

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous